Mailing-List: contact dev-help@ignite.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@ignite.apache.org
MIME-Version: 1.0
In-Reply-To: <bf7622c5-a964-11cd-5e60-2b67a4afd4ea@gmail.com>
References: <CAG7t879O8Dk56WUBN832SEa=sf5mww7UEKj7Z+VjtYz6nf5Bgw@mail.gmail.com>
 <CABuYRcpDPExBKgbcXMb0CMsY7Dnp5uzGyk7CCrauYq8GAmnZ2g@mail.gmail.com> <bf7622c5-a964-11cd-5e60-2b67a4afd4ea@gmail.com>
From: Valentin Kulichenko <valentin.kulichenko@gmail.com>
Date: Wed, 26 Jul 2017 16:48:50 -0700
Message-ID: <CABuYRcoCTswv5gwxt7hk0v92OxZjxqFT4dvfjfEaN9uwd8gx7g@mail.gmail.com>
Subject: Re: ContinuousQueryWithTransformer implementation questions
To: Nikolay Izhikov <nizhikov.dev@gmail.com>, dev@ignite.apache.org
Content-Type: multipart/alternative; boundary="94eb2c0c3d4cc8419f0555411802"
archived-at: Wed, 26 Jul 2017 23:49:32 -0000

--94eb2c0c3d4cc8419f0555411802
Content-Type: text/plain; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable

Yeah, unfortunately current ContinuousQuery object can be used for querying
with transformer. That's actually not good, because adding transformers to
continuous queries and scan queries will be very inconsistent.

AFAIK, there are plans to completely rework query API since we added a lot
of stuff current API is not enough for (DML, DLL, etc.). Probably it makes
sense to consider transformers in the new API as well.

-Val

On Wed, Jul 26, 2017 at 1:32 PM, Nikolay Izhikov <nizhikov.dev@gmail.com>
wrote:

> Hello, Valentin.
>
> As far as I can understand `query(Query<T> qry, IgniteClosure<T, R>
> transformer)` is slightly different from what I should implement.
>
>
> I need to pass two parameter for ContinuousQuery instead of localListener=
:
>
> - Remote Transformer
> - Local Listener for transformed events
>
> and method you provide can accept only transformer.
>
> Moreover I think I should somehow "extend" ContinuousQuery(my proposal is
> new class with similar name) because issue is about possibility of
> optimization of continuous query mechanism.
>
> Thoughts?
>
>
> 26.07.2017 20:56, Valentin Kulichenko =D0=BF=D0=B8=D1=88=D0=B5=D1=82:
>
> Nikolay,
>>
>> We already have the following method for queries with transformer. It
>> currently throws exception for ContinuousQuery.
>>
>> <T, R> QueryCursor<R> query(Query<T> qry, IgniteClosure<T, R> transforme=
r)
>>
>> Would it be possible to utilize it instead of creating new API?
>>
>> -Val
>>
>> On Wed, Jul 26, 2017 at 5:26 AM, =D0=9D=D0=B8=D0=BA=D0=BE=D0=BB=D0=B0=D0=
=B9 =D0=98=D0=B6=D0=B8=D0=BA=D0=BE=D0=B2 <nizhikov.dev@gmail.com>
>> wrote:
>>
>> Hello, Igniters.
>>>
>>> I'm working on IGNITE-425 [1] issue.
>>> I made a couple of changes in my branch [2] so I want to confirm that
>>> changes with community before moving forward:
>>>
>>> Text of issue:
>>>
>>> ```
>>> Currently if updated entry passes the filter, it is sent to node
>>> initiated
>>> the query entirely.
>>> It would be good to provide user with the ability to transform entry an=
d,
>>> for example,
>>> select only fields that are important. This may bring huge economy to
>>> traffic and lower GC pressure as well.
>>> ```
>>>
>>> 1. I create new class ContinuousQueryWithTransformer extends Query:
>>>
>>> Reasons to create entirely new class without extending ContinuousQuery:
>>>
>>>      a. ContinuousQuery is final so user can't extends it. I don't want
>>> to
>>> change that.
>>>      b. ContinuousQuery contains some deprecated
>>> methods(setRemoteFilter) so
>>> with new class we can get rid of them.
>>>      c. Such public API design disallow usage of existing
>>> localEventListener
>>> with new transformedEventListenr in compile time.
>>>
>>> ```
>>>      public final class ContinuousQueryWithTransformer<K, V, T> extends
>>> Query<Cache.Entry<K, V>> {
>>>          public ContinuousQueryWithTransformer<K, V, T>
>>> setRemoteFilterFactory(Factory<? extends CacheEntryEventFilter<K, V>>
>>> rmtFilterFactory) { /**/ }
>>>
>>>          public ContinuousQueryWithTransformer<K, V, T>
>>> setRemoteTransformerFactory(Factory<? extends IgniteBiClosure<K, V, T>>
>>> factory) { /**/ }
>>>
>>>          public ContinuousQueryWithTransformer<K, V, T>
>>> setLocalTransformedEventListener(TransformedEventListener<T>
>>> locTransEvtLsnr) { /**/ }
>>>
>>>          public interface TransformedEventListener<T> {
>>>              void onUpdated(Iterable<? extends T> events) throws
>>> CacheEntryListenerException;
>>>          }
>>>      }
>>> ```
>>>
>>> 2. I want to edit all tests from package
>>> `core/src/test/java/org/apach/ignite/internal/processors/
>>> cache/query/continuous/`
>>> to ensure my implementation fully support existing tests.
>>> I want to make each test can work both for regular ContinousQuery and
>>> ContinuousQueryWithTransformer:
>>>
>>> Existing test:
>>>
>>> ```
>>>          ContinuousQuery<Object, Object> qry =3D new ContinuousQuery<>(=
);
>>>
>>>          qry.setLocalListener(new CacheEntryUpdatedListener<Object,
>>> Object>() {
>>>              @Override public void onUpdated(Iterable<CacheEntryEvent<?=
,
>>> ?>>
>>> evts) {
>>>                  for (CacheEntryEvent evt : evts) {
>>>                      if ((Integer)evt.getValue() >=3D 0)
>>>                          evtCnt.incrementAndGet();
>>>                  }
>>>              }
>>>          });
>>>
>>> ```
>>>
>>> To be:
>>>
>>> ```
>>>          Query qry =3D createContinuousQuery();
>>>
>>>          setLocalListener(qry, new CI1<T2<Object, Object>>() {
>>>              @Override public void apply(T2<Object, Object> e) {
>>>                  if ((Integer)e.getValue() >=3D 0)
>>>                      evtCnt.incrementAndGet();
>>>              }
>>>          });
>>> ```
>>>
>>> Base class to support setLocalListener:
>>>
>>> ```
>>>      protected <K, V> void setLocalListener(Query q, CI1<T2<K, V>>
>>> lsnrClsr)
>>> {
>>>          if (isContinuousWithTransformer()) {
>>>              ((ContinuousQueryWithTransformer)q)
>>>                  .setLocalTransformedEventListener(new
>>> TransformedEventListenerImpl(lsnrClsr));
>>>          } else
>>>              ((ContinuousQuery)q).setLocalListener(new
>>> CacheInvokeListener(lsnrClsr));
>>>      }
>>>
>>>      protected static class CacheInvokeListener<K, V>  {
>>>          private CI1<T2<K, V>> clsr;
>>>
>>>          @Override public void onUpdated(Iterable<CacheEntryEvent<?
>>> extends
>>> K, ? extends V>> events)
>>>              throws CacheEntryListenerException {
>>>              for (CacheEntryEvent<? extends K, ? extends V> e : events)
>>>                  clsr.apply(ignite, new T2<>(e.getKey(), e.getValue()))=
;
>>>          }
>>>      }
>>>
>>>      protected static class TransformedEventListenerImpl<K, V>
>>> implements
>>> TransformedEventListener {
>>>          private IgniteBiInClosure<Ignite, T2<K, V>> clsr;
>>>
>>>          @Override public void onUpdated(Iterable evts) throws
>>> CacheEntryListenerException {
>>>              for (Object e : evts) {
>>>                  clsr.apply((T2)e);
>>>              }
>>>          }
>>>      }
>>> ```
>>>
>>> Thoughts?
>>>
>>> [1] https://issues.apache.org/jira/browse/IGNITE-425
>>> [2] https://github.com/nizhikov/ignite/pull/9/files
>>>
>>> --
>>> Nikolay Izhikov
>>> NIzhikov.dev@gmail.com
>>>
>>>
>>

--94eb2c0c3d4cc8419f0555411802--