Re: joins and low selectivity optimization

[Stamatis Zampetakis]

We did this in the project that I am currently working on but this part is
not open source so I don't have a link to give you.

Στις Σάβ, 23 Φεβ 2019 στις 9:35 μ.μ., ο/η Andrei Sereda 
έγραψε:

> > There are two tricky (but doable) parts:
> > 1) pass bound variables from one convention to the other;
> > 2) push the Filter (associated with the Correlate) holding the condition
> on the correlated variable to elastic;
>
> Do you know if any projects (libraries) already did this ? I would like to
> look at some code examples.
>
> On Fri, Feb 22, 2019 at 6:36 PM Stamatis Zampetakis 
> wrote:
>
> > No need to mention it, I also appreciate your contributions/discussions.
> >
> > Block-based nested loops is not implemented but we already have
> tuple-based
> > nested loop [1] and it is a good starting point.
> >
> > Regarding the particular query that you cited I think that even
> tuple-based
> > nested loop would work really well. For each tuple in orders you will
> bind
> > the productId and you will use it to probe product. Since you have 100
> > tuples in mongo you will end up sending 100 queries in elastic each one
> > though returning a single tuple corresponding to the appropriate id.
> There
> > are two tricky (but doable) parts:
> >
> > 1) pass bound variables from one convention to the other;
> > 2) push the Filter (associated with the Correlate) holding the condition
> on
> > the correlated variable to elastic;
> >
> > Having block-based nested loop could be used to send only one query to
> > Elastic instead of 100. If you start working on a block-based nested loop
> > join I would be happy to help/review the code. In any case, I plan to
> work
> > on it in the following months if I manage to find some time.
> >
> > [1]
> >
> >
> https://github.com/apache/calcite/blob/c3fd74a897ca1b469b6b776baeaa3c660ce5876a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java#L1156
> >
> > Στις Παρ, 22 Φεβ 2019 στις 12:21 π.μ., ο/η Andrei Sereda
>  > >
> > έγραψε:
> >
> > > Hi Stamatis,
> > >
> > > As usual, appreciate your time to answer my questions.
> > >
> > > I will try to give more context below.
> > >
> > > I am not I understand why bloom-filters and block based nested loop are
> > not
> > > possible for your use case
> > >
> > > Our destination data-source (ElasticSearch / Mongo) doesn’t support
> > > filtering based on provided bloom filter and fetching everything in
> > memory
> > > is not an option for us (reference data may have up to 200M records).
> > > Regarding block-based nested loops. From previous Vadym’s message I
> > > understand it is not yet implemented “Support for “batched” correlated
> > > execution is likely not there” (correct me if I’m wrong).
> > >
> > > About physical level implementation. Can you please point me to some
> > > examples of how I would introduce custom EnumerableSemiJoin to perform
> > the
> > > following join (efficiently) :
> > >
> > > select o.quantity, p.price, p.descriptionfrom
> > >   -- Order table is 100 records (in mongo)
> > >   mongo.Order ojoin
> > >   -- Product table is 200M records (in elastic)
> > >   elastic.Product p   on (o.productId = p.id)
> > >
> > > To me, the efficient way to do it would be fetching 100 records from
> > > elastic.Product table first and perform join afterwards. I acknowledge
> > that
> > > I need to better educate myself about calcite physical conventions (any
> > > examples are appreciated).
> > >
> > > The reason I’m introducing streams into picture is because mongo has
> > notion
> > > of change streams  so
> > > potentially mongo table can also be a StreamableTable
> > >  (I would like to submit
> a
> > > separate PR for this). Doing efficient joins for streams still stands.
> > >
> > > Many thanks,
> > > Andrei.
> > >
> > > On Wed, Feb 20, 2019 at 4:23 AM Stamatis Zampetakis  >
> > > wrote:
> > >
> > > > Hi Andrei,
> > > >
> > > > I am not I understand why bloom-filters and block based nested loop
> are
> > > not
> > > > possible for your use case but I will try to provide some answers to
> > the
> > > > new questions you raised.
> > > >
> > > > By adding streams in the discussion I guess you add some additional
> > > > limitations on one side of the join (Orders table):
> > > >
> > > > (i) iterators/enumerators cannot be restarted;
> > > > (ii) iterators/enumerators are infinite;
> > > > (iii) there are no indexes.
> > > >
> > > > Andrei> 1. Am I correct to assume that each event in Orders table
> > (which
> > > is
> > > > a stream) will trigger full table scan (without filter) on Products
> > > table?
> > > >
> > > > The limitations imposed by streams already exclude some join options
> > but
> > > > still it doesn't mean that there is only one way to execute the join.
> > > > So the answer to this question is in general no. Consider for
> instance
> > > that
> > > > the products table is small (and fits in 

Re: joins and low selectivity optimization

[Andrei Sereda]

> There are two tricky (but doable) parts:
> 1) pass bound variables from one convention to the other;
> 2) push the Filter (associated with the Correlate) holding the condition
on the correlated variable to elastic;

Do you know if any projects (libraries) already did this ? I would like to
look at some code examples.

On Fri, Feb 22, 2019 at 6:36 PM Stamatis Zampetakis 
wrote:

> No need to mention it, I also appreciate your contributions/discussions.
>
> Block-based nested loops is not implemented but we already have tuple-based
> nested loop [1] and it is a good starting point.
>
> Regarding the particular query that you cited I think that even tuple-based
> nested loop would work really well. For each tuple in orders you will bind
> the productId and you will use it to probe product. Since you have 100
> tuples in mongo you will end up sending 100 queries in elastic each one
> though returning a single tuple corresponding to the appropriate id. There
> are two tricky (but doable) parts:
>
> 1) pass bound variables from one convention to the other;
> 2) push the Filter (associated with the Correlate) holding the condition on
> the correlated variable to elastic;
>
> Having block-based nested loop could be used to send only one query to
> Elastic instead of 100. If you start working on a block-based nested loop
> join I would be happy to help/review the code. In any case, I plan to work
> on it in the following months if I manage to find some time.
>
> [1]
>
> https://github.com/apache/calcite/blob/c3fd74a897ca1b469b6b776baeaa3c660ce5876a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java#L1156
>
> Στις Παρ, 22 Φεβ 2019 στις 12:21 π.μ., ο/η Andrei Sereda  >
> έγραψε:
>
> > Hi Stamatis,
> >
> > As usual, appreciate your time to answer my questions.
> >
> > I will try to give more context below.
> >
> > I am not I understand why bloom-filters and block based nested loop are
> not
> > possible for your use case
> >
> > Our destination data-source (ElasticSearch / Mongo) doesn’t support
> > filtering based on provided bloom filter and fetching everything in
> memory
> > is not an option for us (reference data may have up to 200M records).
> > Regarding block-based nested loops. From previous Vadym’s message I
> > understand it is not yet implemented “Support for “batched” correlated
> > execution is likely not there” (correct me if I’m wrong).
> >
> > About physical level implementation. Can you please point me to some
> > examples of how I would introduce custom EnumerableSemiJoin to perform
> the
> > following join (efficiently) :
> >
> > select o.quantity, p.price, p.descriptionfrom
> >   -- Order table is 100 records (in mongo)
> >   mongo.Order ojoin
> >   -- Product table is 200M records (in elastic)
> >   elastic.Product p   on (o.productId = p.id)
> >
> > To me, the efficient way to do it would be fetching 100 records from
> > elastic.Product table first and perform join afterwards. I acknowledge
> that
> > I need to better educate myself about calcite physical conventions (any
> > examples are appreciated).
> >
> > The reason I’m introducing streams into picture is because mongo has
> notion
> > of change streams  so
> > potentially mongo table can also be a StreamableTable
> >  (I would like to submit a
> > separate PR for this). Doing efficient joins for streams still stands.
> >
> > Many thanks,
> > Andrei.
> >
> > On Wed, Feb 20, 2019 at 4:23 AM Stamatis Zampetakis 
> > wrote:
> >
> > > Hi Andrei,
> > >
> > > I am not I understand why bloom-filters and block based nested loop are
> > not
> > > possible for your use case but I will try to provide some answers to
> the
> > > new questions you raised.
> > >
> > > By adding streams in the discussion I guess you add some additional
> > > limitations on one side of the join (Orders table):
> > >
> > > (i) iterators/enumerators cannot be restarted;
> > > (ii) iterators/enumerators are infinite;
> > > (iii) there are no indexes.
> > >
> > > Andrei> 1. Am I correct to assume that each event in Orders table
> (which
> > is
> > > a stream) will trigger full table scan (without filter) on Products
> > table?
> > >
> > > The limitations imposed by streams already exclude some join options
> but
> > > still it doesn't mean that there is only one way to execute the join.
> > > So the answer to this question is in general no. Consider for instance
> > that
> > > the products table is small (and fits in memory) then a
> > > hash join algorithm could be used where you build a hash table based on
> > > products and then probe the hash table with the stream relation.
> > >
> > > I would suspect that the default rule set in Calcite chooses this join
> > > option by default (i.e., EnumerableJoin).
> > >
> > > Andrei> 2. Can I register my custom rule to rewrite the query when,
> say,
> > > Orders and Products tables are present to manually add a sub 

Re: joins and low selectivity optimization

[Stamatis Zampetakis]

No need to mention it, I also appreciate your contributions/discussions.

Block-based nested loops is not implemented but we already have tuple-based
nested loop [1] and it is a good starting point.

Regarding the particular query that you cited I think that even tuple-based
nested loop would work really well. For each tuple in orders you will bind
the productId and you will use it to probe product. Since you have 100
tuples in mongo you will end up sending 100 queries in elastic each one
though returning a single tuple corresponding to the appropriate id. There
are two tricky (but doable) parts:

1) pass bound variables from one convention to the other;
2) push the Filter (associated with the Correlate) holding the condition on
the correlated variable to elastic;

Having block-based nested loop could be used to send only one query to
Elastic instead of 100. If you start working on a block-based nested loop
join I would be happy to help/review the code. In any case, I plan to work
on it in the following months if I manage to find some time.

[1]
https://github.com/apache/calcite/blob/c3fd74a897ca1b469b6b776baeaa3c660ce5876a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java#L1156

Στις Παρ, 22 Φεβ 2019 στις 12:21 π.μ., ο/η Andrei Sereda 
έγραψε:

> Hi Stamatis,
>
> As usual, appreciate your time to answer my questions.
>
> I will try to give more context below.
>
> I am not I understand why bloom-filters and block based nested loop are not
> possible for your use case
>
> Our destination data-source (ElasticSearch / Mongo) doesn’t support
> filtering based on provided bloom filter and fetching everything in memory
> is not an option for us (reference data may have up to 200M records).
> Regarding block-based nested loops. From previous Vadym’s message I
> understand it is not yet implemented “Support for “batched” correlated
> execution is likely not there” (correct me if I’m wrong).
>
> About physical level implementation. Can you please point me to some
> examples of how I would introduce custom EnumerableSemiJoin to perform the
> following join (efficiently) :
>
> select o.quantity, p.price, p.descriptionfrom
>   -- Order table is 100 records (in mongo)
>   mongo.Order ojoin
>   -- Product table is 200M records (in elastic)
>   elastic.Product p   on (o.productId = p.id)
>
> To me, the efficient way to do it would be fetching 100 records from
> elastic.Product table first and perform join afterwards. I acknowledge that
> I need to better educate myself about calcite physical conventions (any
> examples are appreciated).
>
> The reason I’m introducing streams into picture is because mongo has notion
> of change streams  so
> potentially mongo table can also be a StreamableTable
>  (I would like to submit a
> separate PR for this). Doing efficient joins for streams still stands.
>
> Many thanks,
> Andrei.
>
> On Wed, Feb 20, 2019 at 4:23 AM Stamatis Zampetakis 
> wrote:
>
> > Hi Andrei,
> >
> > I am not I understand why bloom-filters and block based nested loop are
> not
> > possible for your use case but I will try to provide some answers to the
> > new questions you raised.
> >
> > By adding streams in the discussion I guess you add some additional
> > limitations on one side of the join (Orders table):
> >
> > (i) iterators/enumerators cannot be restarted;
> > (ii) iterators/enumerators are infinite;
> > (iii) there are no indexes.
> >
> > Andrei> 1. Am I correct to assume that each event in Orders table (which
> is
> > a stream) will trigger full table scan (without filter) on Products
> table?
> >
> > The limitations imposed by streams already exclude some join options but
> > still it doesn't mean that there is only one way to execute the join.
> > So the answer to this question is in general no. Consider for instance
> that
> > the products table is small (and fits in memory) then a
> > hash join algorithm could be used where you build a hash table based on
> > products and then probe the hash table with the stream relation.
> >
> > I would suspect that the default rule set in Calcite chooses this join
> > option by default (i.e., EnumerableJoin).
> >
> > Andrei> 2. Can I register my custom rule to rewrite the query when, say,
> > Orders and Products tables are present to manually add a sub query ?
> >
> > It seems that this kind of optimizations are more appropriate at the
> > physical level so if you end up writting custom rules I think it may be
> > better
> > to target a physical convention (e.g., Enumerable relnodes). Moreover, it
> > seems like you want to introduce the equivalent of semi-join reducers so
> > in principle you should introduce something similar to
> EnumerableSemiJoin.
> > I am saying similar because the current implementation of
> > EnumerableSemiJoin
> > is almost the same to the EnumerableJoin (i.e., hash-join based) so I
> would
> > assume that you are not 

Re: joins and low selectivity optimization

[Andrei Sereda]

Hi Stamatis,

As usual, appreciate your time to answer my questions.

I will try to give more context below.

I am not I understand why bloom-filters and block based nested loop are not
possible for your use case

Our destination data-source (ElasticSearch / Mongo) doesn’t support
filtering based on provided bloom filter and fetching everything in memory
is not an option for us (reference data may have up to 200M records).
Regarding block-based nested loops. From previous Vadym’s message I
understand it is not yet implemented “Support for “batched” correlated
execution is likely not there” (correct me if I’m wrong).

About physical level implementation. Can you please point me to some
examples of how I would introduce custom EnumerableSemiJoin to perform the
following join (efficiently) :

select o.quantity, p.price, p.descriptionfrom
  -- Order table is 100 records (in mongo)
  mongo.Order ojoin
  -- Product table is 200M records (in elastic)
  elastic.Product p   on (o.productId = p.id)

To me, the efficient way to do it would be fetching 100 records from
elastic.Product table first and perform join afterwards. I acknowledge that
I need to better educate myself about calcite physical conventions (any
examples are appreciated).

The reason I’m introducing streams into picture is because mongo has notion
of change streams  so
potentially mongo table can also be a StreamableTable
 (I would like to submit a
separate PR for this). Doing efficient joins for streams still stands.

Many thanks,
Andrei.

On Wed, Feb 20, 2019 at 4:23 AM Stamatis Zampetakis 
wrote:

> Hi Andrei,
>
> I am not I understand why bloom-filters and block based nested loop are not
> possible for your use case but I will try to provide some answers to the
> new questions you raised.
>
> By adding streams in the discussion I guess you add some additional
> limitations on one side of the join (Orders table):
>
> (i) iterators/enumerators cannot be restarted;
> (ii) iterators/enumerators are infinite;
> (iii) there are no indexes.
>
> Andrei> 1. Am I correct to assume that each event in Orders table (which is
> a stream) will trigger full table scan (without filter) on Products table?
>
> The limitations imposed by streams already exclude some join options but
> still it doesn't mean that there is only one way to execute the join.
> So the answer to this question is in general no. Consider for instance that
> the products table is small (and fits in memory) then a
> hash join algorithm could be used where you build a hash table based on
> products and then probe the hash table with the stream relation.
>
> I would suspect that the default rule set in Calcite chooses this join
> option by default (i.e., EnumerableJoin).
>
> Andrei> 2. Can I register my custom rule to rewrite the query when, say,
> Orders and Products tables are present to manually add a sub query ?
>
> It seems that this kind of optimizations are more appropriate at the
> physical level so if you end up writting custom rules I think it may be
> better
> to target a physical convention (e.g., Enumerable relnodes). Moreover, it
> seems like you want to introduce the equivalent of semi-join reducers so
> in principle you should introduce something similar to EnumerableSemiJoin.
> I am saying similar because the current implementation of
> EnumerableSemiJoin
> is almost the same to the EnumerableJoin (i.e., hash-join based) so I would
> assume that you are not going to gain match by doing so.
>
> Andrei> 3. Do I have to disable SubQueryRemoveRule in this case ?
>
> If I remember correctly the rule tries to transform correlates to joins so
> I guess it depends on what kind of physical join algorithm you want to use.
> Correlate aka., nested loop join can be a great alternative in some cases
> as other people mentioned before.
>
> Andrei> 4. Vadym, not sure how sub-query computation will work. Can I
> partially execute the query and convert the subquery into EnumerableValues
> ?
>
> Not sure what do you mean here. If you write/choose (by custom rules) the
> algorithm your self you can control exactly what is happening and how the
> query will be executed.
>
> Andrei>Is there a way to solve this problem non-generically ?
> We’re also hitting this limitation in Flink (which uses calcite but not
> calcite streams) for similar use-case.
>
> It seems that you have some concrete query plans (using built-in calcite &
> other operators) which are not performing well.
> Maybe it would be helpful if you could share the problematic plan in this
> discussion.
>
> Best,
> Stamatis
>
>
> Στις Τετ, 20 Φεβ 2019 στις 1:32 π.μ., ο/η Andrei Sereda 
> έγραψε:
>
> > Hello,
> >
> > I would like to resurrect this thread in the context of calcite streams.
> > Unfortunately bloom-filters is not an option for the data-sources being
> > used.
> >
> > Say one has stream to table join
> > 

Re: joins and low selectivity optimization

[Stamatis Zampetakis]

Hi Andrei,

I am not I understand why bloom-filters and block based nested loop are not
possible for your use case but I will try to provide some answers to the
new questions you raised.

By adding streams in the discussion I guess you add some additional
limitations on one side of the join (Orders table):

(i) iterators/enumerators cannot be restarted;
(ii) iterators/enumerators are infinite;
(iii) there are no indexes.

Andrei> 1. Am I correct to assume that each event in Orders table (which is
a stream) will trigger full table scan (without filter) on Products table?

The limitations imposed by streams already exclude some join options but
still it doesn't mean that there is only one way to execute the join.
So the answer to this question is in general no. Consider for instance that
the products table is small (and fits in memory) then a
hash join algorithm could be used where you build a hash table based on
products and then probe the hash table with the stream relation.

I would suspect that the default rule set in Calcite chooses this join
option by default (i.e., EnumerableJoin).

Andrei> 2. Can I register my custom rule to rewrite the query when, say,
Orders and Products tables are present to manually add a sub query ?

It seems that this kind of optimizations are more appropriate at the
physical level so if you end up writting custom rules I think it may be
better
to target a physical convention (e.g., Enumerable relnodes). Moreover, it
seems like you want to introduce the equivalent of semi-join reducers so
in principle you should introduce something similar to EnumerableSemiJoin.
I am saying similar because the current implementation of EnumerableSemiJoin
is almost the same to the EnumerableJoin (i.e., hash-join based) so I would
assume that you are not going to gain match by doing so.

Andrei> 3. Do I have to disable SubQueryRemoveRule in this case ?

If I remember correctly the rule tries to transform correlates to joins so
I guess it depends on what kind of physical join algorithm you want to use.
Correlate aka., nested loop join can be a great alternative in some cases
as other people mentioned before.

Andrei> 4. Vadym, not sure how sub-query computation will work. Can I
partially execute the query and convert the subquery into EnumerableValues ?

Not sure what do you mean here. If you write/choose (by custom rules) the
algorithm your self you can control exactly what is happening and how the
query will be executed.

Andrei>Is there a way to solve this problem non-generically ?
We’re also hitting this limitation in Flink (which uses calcite but not
calcite streams) for similar use-case.

It seems that you have some concrete query plans (using built-in calcite &
other operators) which are not performing well.
Maybe it would be helpful if you could share the problematic plan in this
discussion.

Best,
Stamatis


Στις Τετ, 20 Φεβ 2019 στις 1:32 π.μ., ο/η Andrei Sereda 
έγραψε:

> Hello,
>
> I would like to resurrect this thread in the context of calcite streams.
> Unfortunately bloom-filters is not an option for the data-sources being
> used.
>
> Say one has stream to table join
> .
> From docs example:
>
> SELECT STREAM
> o.productId, o.orderId, o.units, p.name, p.unitPrice FROM Orders
> AS o -- streamable Table JOIN Products AS p -- reference data table
> ON o.productId = p.productId;
>
>
>1. Am I correct to assume that each event in Orders table (which is a
>stream) will trigger full table scan (without filter) on Products table
>?
>2. Can I register my custom rule to rewrite the query when, say, Orders
>and Products tables are present to manually add a sub query ?
>3. Do I have to disable SubQueryRemoveRule in this case ?
>4. Vadym, not sure how sub-query computation will work. Can I partially
>execute the query and convert the subquery into EnumerableValues ?
>
> Is there a way to solve this problem non-generically ?
>
> We’re also hitting this limitation in Flink (which uses calcite but not
> calcite streams) for similar use-case.
>
> Many Thanks,
> Andrei.
>
> On Thu, Aug 30, 2018 at 5:27 PM Vineet Garg  wrote:
>
> > Hive actually does this optimization (it is called semi-join reduction)
> by
> > generating bloom-filters on one side and passing it on to the other side.
> > This is not a rewrite but instead a physical implementation.
> >
> > Vineet
> >
> > On Aug 29, 2018, at 10:34 AM, Vladimir Sitnikov <
> > sitnikov.vladi...@gmail.com> wrote:
> >
> > Nested loops are never likely to happe
> >
> > What's wrong with that?
> >
> > Apparently Andrei asks for that, and "subquery precomputation" is quite
> > close to nested loops in my opinion.
> >
> > Vladimir
> >
> >
>

Re: joins and low selectivity optimization

[Andrei Sereda]

Hello,

I would like to resurrect this thread in the context of calcite streams.
Unfortunately bloom-filters is not an option for the data-sources being
used.

Say one has stream to table join
.
>From docs example:

SELECT STREAM
o.productId, o.orderId, o.units, p.name, p.unitPrice FROM Orders
AS o -- streamable Table JOIN Products AS p -- reference data table
ON o.productId = p.productId;


   1. Am I correct to assume that each event in Orders table (which is a
   stream) will trigger full table scan (without filter) on Products table
   ?
   2. Can I register my custom rule to rewrite the query when, say, Orders
   and Products tables are present to manually add a sub query ?
   3. Do I have to disable SubQueryRemoveRule in this case ?
   4. Vadym, not sure how sub-query computation will work. Can I partially
   execute the query and convert the subquery into EnumerableValues ?

Is there a way to solve this problem non-generically ?

We’re also hitting this limitation in Flink (which uses calcite but not
calcite streams) for similar use-case.

Many Thanks,
Andrei.

On Thu, Aug 30, 2018 at 5:27 PM Vineet Garg  wrote:

> Hive actually does this optimization (it is called semi-join reduction) by
> generating bloom-filters on one side and passing it on to the other side.
> This is not a rewrite but instead a physical implementation.
>
> Vineet
>
> On Aug 29, 2018, at 10:34 AM, Vladimir Sitnikov <
> sitnikov.vladi...@gmail.com> wrote:
>
> Nested loops are never likely to happe
>
> What's wrong with that?
>
> Apparently Andrei asks for that, and "subquery precomputation" is quite
> close to nested loops in my opinion.
>
> Vladimir
>
>

Re: joins and low selectivity optimization

[Vineet Garg]

Hive actually does this optimization (it is called semi-join reduction) by 
generating bloom-filters on one side and passing it on to the other side.
This is not a rewrite but instead a physical implementation.

Vineet

On Aug 29, 2018, at 10:34 AM, Vladimir Sitnikov 
mailto:sitnikov.vladi...@gmail.com>> wrote:

Nested loops are never likely to happe

What's wrong with that?

Apparently Andrei asks for that, and "subquery precomputation" is quite
close to nested loops in my opinion.

Vladimir

Re: joins and low selectivity optimization

[Vladimir Sitnikov]

>Nested loops are never likely to happe

What's wrong with that?

Apparently Andrei asks for that, and "subquery precomputation" is quite
close to nested loops in my opinion.

Vladimir

Re: joins and low selectivity optimization

[Julian Hyde]

Regarding Vladimir’s ideas of Bloom filters and nested loop joins. Both are 
excellent if you can do them. They are fairly easy in single-node architectures 
(especially single-threaded) but get harder in distributed architectures. Bloom 
filters (also magic sets) require data to be pushed “up stream”, and may 
require re-starting sub-graphs.

So, you have to devise query processing algorithms that are appropriate for 
your architecture.

Hive is an example of a highly distributed, parallel engine. Hive would like to 
do Bloom filters but has still not gotten around to it. Nested loops are never 
likely to happen. But Hive uses other techniques.

Julian


> On Aug 29, 2018, at 8:37 AM, Andrei Sereda  wrote:
> 
> Hi Vladimir,
> 
> Thanks for follow-up and explanation. I wanted to make sure I'm not missing
> (mis-understanding) anything.
> 
> Andrei.
> 
> 
> 
> On Wed, Aug 29, 2018 at 11:01 AM Vladimir Sitnikov <
> sitnikov.vladi...@gmail.com> wrote:
> 
>> One of the approaches to such queries is to throw Bloom filters all over
>> the place.
>> 
>> That is it could execute "small side" of the join, collect the ids (or a
>> lossy version of it in a form of Bloom filters),
>> and it could propagate that Bloom filter to the second source to reduce the
>> set of rows produced by the second row source.
>> Then the join would be easier to do since the second row source is reduced.
>> 
>> The sad thing is not all systems support propagation of bloom filters.
>> 
>>> select *from
>>> t1 join t2 on (t1.id = t2.id)where
>>> t2.id in (select id from t1) -- force sub selec
>> 
>> What if Calcite did just a regular batched nested loop join?
>> That is:
>> 1. Fetch next 10 rows from t1
>> 2. Fetch "from t2 where id in (...)"
>> 3. goto 1
>> 
>> It can be expressed via correlated subqueries, however:
>> a) I'm not sure correlated subqueries work great at the moment
>> b) Support for "batched" correlated execution is likely not there
>> c) Calcite should somehow know the true cost of "from t2 where id in (1,2)"
>> vs "from t2 where id in (1,2,3,4)". In other words, current costing model
>> does not take into account if the table has index or not. One can code such
>> costing rules, however I think it is not there yet.
>> 
>> Vladimir
>> 

Re: joins and low selectivity optimization

[Andrei Sereda]

Hi Vladimir,

Thanks for follow-up and explanation. I wanted to make sure I'm not missing
(mis-understanding) anything.

Andrei.



On Wed, Aug 29, 2018 at 11:01 AM Vladimir Sitnikov <
sitnikov.vladi...@gmail.com> wrote:

> One of the approaches to such queries is to throw Bloom filters all over
> the place.
>
> That is it could execute "small side" of the join, collect the ids (or a
> lossy version of it in a form of Bloom filters),
> and it could propagate that Bloom filter to the second source to reduce the
> set of rows produced by the second row source.
> Then the join would be easier to do since the second row source is reduced.
>
> The sad thing is not all systems support propagation of bloom filters.
>
> >select *from
> >  t1 join t2 on (t1.id = t2.id)where
> >  t2.id in (select id from t1) -- force sub selec
>
> What if Calcite did just a regular batched nested loop join?
> That is:
> 1. Fetch next 10 rows from t1
> 2. Fetch "from t2 where id in (...)"
> 3. goto 1
>
> It can be expressed via correlated subqueries, however:
> a) I'm not sure correlated subqueries work great at the moment
> b) Support for "batched" correlated execution is likely not there
> c) Calcite should somehow know the true cost of "from t2 where id in (1,2)"
> vs "from t2 where id in (1,2,3,4)". In other words, current costing model
> does not take into account if the table has index or not. One can code such
> costing rules, however I think it is not there yet.
>
> Vladimir
>

Re: joins and low selectivity optimization

[Vladimir Sitnikov]

One of the approaches to such queries is to throw Bloom filters all over
the place.

That is it could execute "small side" of the join, collect the ids (or a
lossy version of it in a form of Bloom filters),
and it could propagate that Bloom filter to the second source to reduce the
set of rows produced by the second row source.
Then the join would be easier to do since the second row source is reduced.

The sad thing is not all systems support propagation of bloom filters.

>select *from
>  t1 join t2 on (t1.id = t2.id)where
>  t2.id in (select id from t1) -- force sub selec

What if Calcite did just a regular batched nested loop join?
That is:
1. Fetch next 10 rows from t1
2. Fetch "from t2 where id in (...)"
3. goto 1

It can be expressed via correlated subqueries, however:
a) I'm not sure correlated subqueries work great at the moment
b) Support for "batched" correlated execution is likely not there
c) Calcite should somehow know the true cost of "from t2 where id in (1,2)"
vs "from t2 where id in (1,2,3,4)". In other words, current costing model
does not take into account if the table has index or not. One can code such
costing rules, however I think it is not there yet.

Vladimir

Re: joins and low selectivity optimization

[Andrei Sereda]

A follow-up question. I presume this is applicable equally to joins without
predicates ?

select *from
  t1 join t2 on (t1.id = t2.id)

Even if t1 has a much lower cardinality than t2 (say many orders of
magnitude), Calcite would not know to convert t2 query to:

select * from t2 where id in (1, 2, 3) -- many ids

“Manual” optimization

Even the following might get “de-optimized” ?

select *from
  t1 join t2 on (t1.id = t2.id)where
  t2.id in (select id from t1) -- force sub select


On Tue, Aug 28, 2018 at 10:24 PM Julian Hyde  wrote:

> Sure, Calcite makes use of stats in its cost formulas. And you are correct
> that “metadata” is what Calcite calls statistics.
>
> But you have to be careful to only treat statistics as approximate. If the
> statistics were gathered using an “ANALYZE TABLE” command a month ago they
> may be out of date, so you cannot use them to, say, remove “WHERE x < 10”
> if a month ago x only had values 2, 4, and 6.
>
> > On Aug 28, 2018, at 7:14 PM, Andrei Sereda  wrote:
> >
> > Thank you, Michael and Julian, for your answers.
> >
> > Even if optimizers don't have access to data can they have access to
> table
> > statistics ? If I remember correctly Oracle CBO is estimating selectivity
> > based on column distribution (histograms) and some formulas for density
> > . I realize these
> > statistics are not available for all data stores but can calcite
> optimizer
> > be "smarter" when this data is available ?
> >
> > On Tue, Aug 28, 2018 at 9:46 PM Julian Hyde  wrote:
> >
> >> If I recall correctly, Hive does this kind of optimization. It’s pretty
> >> important you have a date dimension table and your fact table is
> >> partitioned on date. Example:
> >>
> >>  select *
> >>  from sales
> >>join date_dim on sales.date_id = date_dim.id
> >>  where sales.product_name = ‘foo'
> >>  and date_dim.quarter = ‘2018-Q2'
> >>
> >> Hive would like to transform it to
> >>
> >>  select *
> >>  from sales
> >>  where date_id in (20180401, 20180402, … , 20180630)
> >>  and sales.product_name = ‘foo'
> >>
> >> by pre-evaluating the query on the date_dim table. It doesn’t do the
> >> optimization at logical planning time (where Calcite is involved) but at
> >> physical planning time (which occurs later). The list of date_id values
> >> allows it to scan a much more limited set of partitions of the sales
> fact
> >> table.
> >>
> >> Michael is correct that optimizers don’t usually have access to data.
> But
> >> if the date_dim table changes only slowly, you could set up a “tripwire”
> >> that will invalidate the plan if the date_dim table happens to change
> >> between planning and execution.
> >>
> >> Julian
> >>
> >>
> >>
> >>
> >>> On Aug 28, 2018, at 6:04 PM, Michael Mior  wrote:
> >>>
> >>> As far as I am aware, the optimizer has no access to data, only
> metadata.
> >>> The traditional way to solve such problems would be to select among
> >>> different join algorithms which perform better for varying
> cardinalities
> >> of
> >>> each side of the join. Unfortunately, I think you're likely to have a
> >> tough
> >>> time extracting the necessary data to do the rewrite you're aiming for.
> >>>
> >>> --
> >>> Michael Mior
> >>> mm...@apache.org
> >>>
> >>>
> >>>
> >>> Le mar. 28 août 2018 à 20:34, Andrei Sereda  a
> écrit :
> >>>
>  Hello,
> 
>  I’m looking for a way to improve performance of a join query.
> 
>  Suppose one joins two heterogeneous sources t1 and t2 with some
> >> predicates.
> 
>  Further assume that cardinality of one of the predicates is very low
>  (compared cardinality of the second one). (How) Is it possible to
> >> convert
>  second query (predicate) to include results (primary keys) of the
> first
> >> one
>  (with low selectivity) ?
>  Example
> 
>  select *from
>  t1 left join t1 on (t1.id = t2.id)where
>  t1.attr = 'foo' and t2.attr = 'bar'
> 
>  Let’s say that predicate t1.attr = 'foo' results in 3 rows (id=1, 2,
> 3).
>  Will it be possible to rewrite t2 query to :
> 
>  select *from t2 where
>   id in (1, 2, 3) and t2.attr = 'bar'
> 
>  I’m aware of existence of Metadata
>  <
> 
> >>
> https://calcite.apache.org/apidocs/org/apache/calcite/rel/metadata/Metadata.html
> >
>  but not sure to use it.
> 
>  Any hits / directions are appreciated.
> 
>  Thanks,
>  Andrei.
> 
> >>
> >>
>
>

Re: joins and low selectivity optimization

[Julian Hyde]

Sure, Calcite makes use of stats in its cost formulas. And you are correct that 
“metadata” is what Calcite calls statistics.

But you have to be careful to only treat statistics as approximate. If the 
statistics were gathered using an “ANALYZE TABLE” command a month ago they may 
be out of date, so you cannot use them to, say, remove “WHERE x < 10” if a 
month ago x only had values 2, 4, and 6.

> On Aug 28, 2018, at 7:14 PM, Andrei Sereda  wrote:
> 
> Thank you, Michael and Julian, for your answers.
> 
> Even if optimizers don't have access to data can they have access to table
> statistics ? If I remember correctly Oracle CBO is estimating selectivity
> based on column distribution (histograms) and some formulas for density
> . I realize these
> statistics are not available for all data stores but can calcite optimizer
> be "smarter" when this data is available ?
> 
> On Tue, Aug 28, 2018 at 9:46 PM Julian Hyde  wrote:
> 
>> If I recall correctly, Hive does this kind of optimization. It’s pretty
>> important you have a date dimension table and your fact table is
>> partitioned on date. Example:
>> 
>>  select *
>>  from sales
>>join date_dim on sales.date_id = date_dim.id
>>  where sales.product_name = ‘foo'
>>  and date_dim.quarter = ‘2018-Q2'
>> 
>> Hive would like to transform it to
>> 
>>  select *
>>  from sales
>>  where date_id in (20180401, 20180402, … , 20180630)
>>  and sales.product_name = ‘foo'
>> 
>> by pre-evaluating the query on the date_dim table. It doesn’t do the
>> optimization at logical planning time (where Calcite is involved) but at
>> physical planning time (which occurs later). The list of date_id values
>> allows it to scan a much more limited set of partitions of the sales fact
>> table.
>> 
>> Michael is correct that optimizers don’t usually have access to data. But
>> if the date_dim table changes only slowly, you could set up a “tripwire”
>> that will invalidate the plan if the date_dim table happens to change
>> between planning and execution.
>> 
>> Julian
>> 
>> 
>> 
>> 
>>> On Aug 28, 2018, at 6:04 PM, Michael Mior  wrote:
>>> 
>>> As far as I am aware, the optimizer has no access to data, only metadata.
>>> The traditional way to solve such problems would be to select among
>>> different join algorithms which perform better for varying cardinalities
>> of
>>> each side of the join. Unfortunately, I think you're likely to have a
>> tough
>>> time extracting the necessary data to do the rewrite you're aiming for.
>>> 
>>> --
>>> Michael Mior
>>> mm...@apache.org
>>> 
>>> 
>>> 
>>> Le mar. 28 août 2018 à 20:34, Andrei Sereda  a écrit :
>>> 
 Hello,
 
 I’m looking for a way to improve performance of a join query.
 
 Suppose one joins two heterogeneous sources t1 and t2 with some
>> predicates.
 
 Further assume that cardinality of one of the predicates is very low
 (compared cardinality of the second one). (How) Is it possible to
>> convert
 second query (predicate) to include results (primary keys) of the first
>> one
 (with low selectivity) ?
 Example
 
 select *from
 t1 left join t1 on (t1.id = t2.id)where
 t1.attr = 'foo' and t2.attr = 'bar'
 
 Let’s say that predicate t1.attr = 'foo' results in 3 rows (id=1, 2, 3).
 Will it be possible to rewrite t2 query to :
 
 select *from t2 where
  id in (1, 2, 3) and t2.attr = 'bar'
 
 I’m aware of existence of Metadata
 <
 
>> https://calcite.apache.org/apidocs/org/apache/calcite/rel/metadata/Metadata.html
> 
 but not sure to use it.
 
 Any hits / directions are appreciated.
 
 Thanks,
 Andrei.
 
>> 
>> 

Re: joins and low selectivity optimization

[Andrei Sereda]

Thank you, Michael and Julian, for your answers.

Even if optimizers don't have access to data can they have access to table
statistics ? If I remember correctly Oracle CBO is estimating selectivity
based on column distribution (histograms) and some formulas for density
. I realize these
statistics are not available for all data stores but can calcite optimizer
be "smarter" when this data is available ?

On Tue, Aug 28, 2018 at 9:46 PM Julian Hyde  wrote:

> If I recall correctly, Hive does this kind of optimization. It’s pretty
> important you have a date dimension table and your fact table is
> partitioned on date. Example:
>
>   select *
>   from sales
> join date_dim on sales.date_id = date_dim.id
>   where sales.product_name = ‘foo'
>   and date_dim.quarter = ‘2018-Q2'
>
> Hive would like to transform it to
>
>   select *
>   from sales
>   where date_id in (20180401, 20180402, … , 20180630)
>   and sales.product_name = ‘foo'
>
> by pre-evaluating the query on the date_dim table. It doesn’t do the
> optimization at logical planning time (where Calcite is involved) but at
> physical planning time (which occurs later). The list of date_id values
> allows it to scan a much more limited set of partitions of the sales fact
> table.
>
> Michael is correct that optimizers don’t usually have access to data. But
> if the date_dim table changes only slowly, you could set up a “tripwire”
> that will invalidate the plan if the date_dim table happens to change
> between planning and execution.
>
> Julian
>
>
>
>
> > On Aug 28, 2018, at 6:04 PM, Michael Mior  wrote:
> >
> > As far as I am aware, the optimizer has no access to data, only metadata.
> > The traditional way to solve such problems would be to select among
> > different join algorithms which perform better for varying cardinalities
> of
> > each side of the join. Unfortunately, I think you're likely to have a
> tough
> > time extracting the necessary data to do the rewrite you're aiming for.
> >
> > --
> > Michael Mior
> > mm...@apache.org
> >
> >
> >
> > Le mar. 28 août 2018 à 20:34, Andrei Sereda  a écrit :
> >
> >> Hello,
> >>
> >> I’m looking for a way to improve performance of a join query.
> >>
> >> Suppose one joins two heterogeneous sources t1 and t2 with some
> predicates.
> >>
> >> Further assume that cardinality of one of the predicates is very low
> >> (compared cardinality of the second one). (How) Is it possible to
> convert
> >> second query (predicate) to include results (primary keys) of the first
> one
> >> (with low selectivity) ?
> >> Example
> >>
> >> select *from
> >>  t1 left join t1 on (t1.id = t2.id)where
> >>  t1.attr = 'foo' and t2.attr = 'bar'
> >>
> >> Let’s say that predicate t1.attr = 'foo' results in 3 rows (id=1, 2, 3).
> >> Will it be possible to rewrite t2 query to :
> >>
> >> select *from t2 where
> >>   id in (1, 2, 3) and t2.attr = 'bar'
> >>
> >> I’m aware of existence of Metadata
> >> <
> >>
> https://calcite.apache.org/apidocs/org/apache/calcite/rel/metadata/Metadata.html
> >>>
> >> but not sure to use it.
> >>
> >> Any hits / directions are appreciated.
> >>
> >> Thanks,
> >> Andrei.
> >>
>
>

Re: joins and low selectivity optimization

[Julian Hyde]

If I recall correctly, Hive does this kind of optimization. It’s pretty 
important you have a date dimension table and your fact table is partitioned on 
date. Example:

  select *
  from sales
join date_dim on sales.date_id = date_dim.id
  where sales.product_name = ‘foo'
  and date_dim.quarter = ‘2018-Q2'

Hive would like to transform it to

  select *
  from sales
  where date_id in (20180401, 20180402, … , 20180630)
  and sales.product_name = ‘foo'

by pre-evaluating the query on the date_dim table. It doesn’t do the 
optimization at logical planning time (where Calcite is involved) but at 
physical planning time (which occurs later). The list of date_id values allows 
it to scan a much more limited set of partitions of the sales fact table.

Michael is correct that optimizers don’t usually have access to data. But if 
the date_dim table changes only slowly, you could set up a “tripwire” that will 
invalidate the plan if the date_dim table happens to change between planning 
and execution.

Julian




> On Aug 28, 2018, at 6:04 PM, Michael Mior  wrote:
> 
> As far as I am aware, the optimizer has no access to data, only metadata.
> The traditional way to solve such problems would be to select among
> different join algorithms which perform better for varying cardinalities of
> each side of the join. Unfortunately, I think you're likely to have a tough
> time extracting the necessary data to do the rewrite you're aiming for.
> 
> --
> Michael Mior
> mm...@apache.org
> 
> 
> 
> Le mar. 28 août 2018 à 20:34, Andrei Sereda  a écrit :
> 
>> Hello,
>> 
>> I’m looking for a way to improve performance of a join query.
>> 
>> Suppose one joins two heterogeneous sources t1 and t2 with some predicates.
>> 
>> Further assume that cardinality of one of the predicates is very low
>> (compared cardinality of the second one). (How) Is it possible to convert
>> second query (predicate) to include results (primary keys) of the first one
>> (with low selectivity) ?
>> Example
>> 
>> select *from
>>  t1 left join t1 on (t1.id = t2.id)where
>>  t1.attr = 'foo' and t2.attr = 'bar'
>> 
>> Let’s say that predicate t1.attr = 'foo' results in 3 rows (id=1, 2, 3).
>> Will it be possible to rewrite t2 query to :
>> 
>> select *from t2 where
>>   id in (1, 2, 3) and t2.attr = 'bar'
>> 
>> I’m aware of existence of Metadata
>> <
>> https://calcite.apache.org/apidocs/org/apache/calcite/rel/metadata/Metadata.html
>>> 
>> but not sure to use it.
>> 
>> Any hits / directions are appreciated.
>> 
>> Thanks,
>> Andrei.
>> 

Re: joins and low selectivity optimization

[Michael Mior]

As far as I am aware, the optimizer has no access to data, only metadata.
The traditional way to solve such problems would be to select among
different join algorithms which perform better for varying cardinalities of
each side of the join. Unfortunately, I think you're likely to have a tough
time extracting the necessary data to do the rewrite you're aiming for.

--
Michael Mior
mm...@apache.org



Le mar. 28 août 2018 à 20:34, Andrei Sereda  a écrit :

> Hello,
>
> I’m looking for a way to improve performance of a join query.
>
> Suppose one joins two heterogeneous sources t1 and t2 with some predicates.
>
> Further assume that cardinality of one of the predicates is very low
> (compared cardinality of the second one). (How) Is it possible to convert
> second query (predicate) to include results (primary keys) of the first one
> (with low selectivity) ?
> Example
>
> select *from
>   t1 left join t1 on (t1.id = t2.id)where
>   t1.attr = 'foo' and t2.attr = 'bar'
>
> Let’s say that predicate t1.attr = 'foo' results in 3 rows (id=1, 2, 3).
> Will it be possible to rewrite t2 query to :
>
> select *from t2 where
>id in (1, 2, 3) and t2.attr = 'bar'
>
> I’m aware of existence of Metadata
> <
> https://calcite.apache.org/apidocs/org/apache/calcite/rel/metadata/Metadata.html
> >
> but not sure to use it.
>
> Any hits / directions are appreciated.
>
> Thanks,
> Andrei.
>

joins and low selectivity optimization

[Andrei Sereda]

Hello,

I’m looking for a way to improve performance of a join query.

Suppose one joins two heterogeneous sources t1 and t2 with some predicates.

Further assume that cardinality of one of the predicates is very low
(compared cardinality of the second one). (How) Is it possible to convert
second query (predicate) to include results (primary keys) of the first one
(with low selectivity) ?
Example

select *from
  t1 left join t1 on (t1.id = t2.id)where
  t1.attr = 'foo' and t2.attr = 'bar'

Let’s say that predicate t1.attr = 'foo' results in 3 rows (id=1, 2, 3).
Will it be possible to rewrite t2 query to :

select *from t2 where
   id in (1, 2, 3) and t2.attr = 'bar'

I’m aware of existence of Metadata

but not sure to use it.

Any hits / directions are appreciated.

Thanks,
Andrei.