ignite-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Kevin Jin (JIRA)" <j...@apache.org>
Subject [jira] [Updated] (IGNITE-7849) Generating a CacheEntryEventFilter from a SqlQuery
Date Wed, 28 Feb 2018 17:06:00 GMT

     [ https://issues.apache.org/jira/browse/IGNITE-7849?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]

Kevin Jin updated IGNITE-7849:
------------------------------
    Description: 
Currently when we want to use the same predicate for the continuous query and initial query,
it's easy enough to write something like this, assuming we are fine with the performance of
ScanQuery:

{code:title=IgniteContinuousQueryExample.java|borderStyle=solid}
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.query.ContinuousQuery;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.query.ScanQuery;
import javax.cache.Cache;
import javax.cache.event.CacheEntryEvent;

public class IgniteContinuousQueryExample {
    private static final String CLUSTER = "TESTGRID";
    private static final String TABLE = "TESTTABLE";

    private static boolean isInteresting(Integer key, String value) {
        return key > 10;
    }

    private static boolean isInteresting(CacheEntryEvent<? extends Integer, ? extends String>
event) {
        return isInteresting(event.getKey(), event.getValue());
    }

    private static void handleEntry(Cache.Entry<? extends Integer, ? extends String>
event) {
        System.out.println("Received value for " + event.getKey() + ": " + event.getValue());
    }

    private static void handleExistingEntries(Iterable<? extends Cache.Entry<? extends
Integer, ? extends String>> events) {
        events.forEach(IgniteContinuousQueryExample::handleEntry);
    }

    public static void main(String[] args) throws InterruptedException {
        try (Ignite client = Ignition.ignite(CLUSTER); IgniteCache<Integer, String>
cache = client.cache(TABLE)) {
            for (int i = 0; i < 20; i++)
                cache.put(i, Integer.toString(i));

            ContinuousQuery<Integer, String> query = new ContinuousQuery<>();
            query.setRemoteFilterFactory(() -> IgniteContinuousQueryExample::isInteresting);
            query.setInitialQuery(new ScanQuery<>(IgniteContinuousQueryExample::isInteresting));

            query.setLocalListener(IgniteContinuousQueryExample::handleExistingEntries);
            try (QueryCursor<Cache.Entry<Integer, String>> resultSet = cache.query(query))
{
                handleExistingEntries(resultSet);

                // Local listener callbacks will no longer be received once resultSet is closed
so
                //  these updates have to be inside the try-with-resources block and we have
to wait
                //  to close resultSet after the final update.
                for (int i = 20; i < 30; i++)
                    cache.put(i, Integer.toString(i));

                Thread.sleep(60000);
            }
        }
    }
}
{code}

However, this becomes more inconvenient when we want to use SqlQuery in the initial query
to take advantage of indexing. This is the best that I can do:

query.setRemoteFilterFactory(() -> entry -> entry.getKey() > 10);
 query.setInitialQuery(new SqlQuery<>(String.sql, "`_key` > 10"));

This is obviously not ideal because we have to specify the predicate in two different ways.
A quick Google revealed that there are products out there that more seamlessly support this
use case of continuous querying. I understand that Ignite isn't built on top of SQL, unlike
the commercial RDBMSes I found, so maybe this is an out-of-scope feature.

  was:
Currently when we want to use the same predicate for the continuous query and initial query,
it's easy enough to write something like this, assuming we are fine with the performance of
ScanQuery:

{{import org.apache.ignite.Ignite;}}
 {{import org.apache.ignite.IgniteCache;}}
 {{import org.apache.ignite.Ignition;}}
 {{import org.apache.ignite.cache.query.ContinuousQuery;}}
 {{import org.apache.ignite.cache.query.QueryCursor;}}
 {{import org.apache.ignite.cache.query.ScanQuery;}}
 {{import javax.cache.Cache;}}
 {{import javax.cache.event.CacheEntryEvent;}}
 {{public class IgniteContinuousQueryExample {}}
   \{{private static final String CLUSTER = "TESTGRID";}}
   \{{private static final String TABLE = "TESTTABLE";}}

{{  }}{{private static boolean isInteresting(Integer key, String value) {}}
 \{{     return key > 10;}}
 {

{   }}}
 {{  private static boolean isInteresting(CacheEntryEvent<? extends Integer, ? extends
String> event) {}}
 \{{     return isInteresting(event.getKey(), event.getValue());}}
 \{{   }

}}
 {{  private static void handleEntry(Cache.Entry<? extends Integer, ? extends String>
event) {}}
 \{{     System.out.println("Received value for " + event.getKey() + ": " + event.getValue());}}
 {

{   }}}
 
 {{  }}{{private static void handleResultSet(Iterable<? extends Cache.Entry<? extends
Integer, ? extends String>> events) {}}
 \{{     events.forEach(IgniteContinuousQueryExample::handleEntry);}}
 \{{   }

}}
 {{  public static void main(String[] args) throws InterruptedException {}}
 \{{     try (Ignite client = Ignition.ignite(CLUSTER); IgniteCache<Integer, String>
cache = client.cache(TABLE)) {}}
 \{{       for (int i = 0; i < 20; i++)}}
 \{{         cache.put(i, Integer.toString());}}

{{      }}{{ContinuousQuery<Integer, String> query = new ContinuousQuery<>();}}
 \{{       query.setRemoteFilterFactory(() -> IgniteContinuousQueryExample::isInteresting);}}
 \{{       query.setInitialQuery(new ScanQuery<>(IgniteContinuousQueryExample::isInteresting));}}

{{      }}{{query.setLocalListener(IgniteContinuousQueryExample::handleResultSet);}}
 \{{       try (QueryCursor<Cache.Entry<Integer, String>> resultSet = cache.query(query))
{}}
 \{{         handleResultSet(resultSet);}}

{{        for (int i = 20; i < 30; i++)}}
 \{{           cache.put(i, Integer.toString());}}{{Thread.sleep(60000);}}
 {

{       }

}}
 {

{     }

}}
 {

{   }

}}
 {{}}}

However, this becomes more inconvenient when we want to use SqlQuery in the initial query
to take advantage of indexing. This is the best that I can do:

query.setRemoteFilterFactory(() -> entry -> entry.getKey() > 10);
 query.setInitialQuery(new SqlQuery<>(String.sql, "`_key` > 10"));

This is obviously not ideal because we have to specify the predicate in two different ways.
A quick Google revealed that there are products out there that more seamlessly support this
use case of continuous querying. I understand that Ignite isn't built on top of SQL, unlike
the commercial RDBMSes I found, so maybe this is an out-of-scope feature.


> Generating a CacheEntryEventFilter from a SqlQuery
> --------------------------------------------------
>
>                 Key: IGNITE-7849
>                 URL: https://issues.apache.org/jira/browse/IGNITE-7849
>             Project: Ignite
>          Issue Type: Wish
>          Components: cache, sql
>    Affects Versions: 2.3
>            Reporter: Kevin Jin
>            Priority: Trivial
>              Labels: features, usability
>             Fix For: None
>
>   Original Estimate: 0.2h
>  Remaining Estimate: 0.2h
>
> Currently when we want to use the same predicate for the continuous query and initial
query, it's easy enough to write something like this, assuming we are fine with the performance
of ScanQuery:
> {code:title=IgniteContinuousQueryExample.java|borderStyle=solid}
> import org.apache.ignite.Ignite;
> import org.apache.ignite.IgniteCache;
> import org.apache.ignite.Ignition;
> import org.apache.ignite.cache.query.ContinuousQuery;
> import org.apache.ignite.cache.query.QueryCursor;
> import org.apache.ignite.cache.query.ScanQuery;
> import javax.cache.Cache;
> import javax.cache.event.CacheEntryEvent;
> public class IgniteContinuousQueryExample {
>     private static final String CLUSTER = "TESTGRID";
>     private static final String TABLE = "TESTTABLE";
>     private static boolean isInteresting(Integer key, String value) {
>         return key > 10;
>     }
>     private static boolean isInteresting(CacheEntryEvent<? extends Integer, ? extends
String> event) {
>         return isInteresting(event.getKey(), event.getValue());
>     }
>     private static void handleEntry(Cache.Entry<? extends Integer, ? extends String>
event) {
>         System.out.println("Received value for " + event.getKey() + ": " + event.getValue());
>     }
>     private static void handleExistingEntries(Iterable<? extends Cache.Entry<?
extends Integer, ? extends String>> events) {
>         events.forEach(IgniteContinuousQueryExample::handleEntry);
>     }
>     public static void main(String[] args) throws InterruptedException {
>         try (Ignite client = Ignition.ignite(CLUSTER); IgniteCache<Integer, String>
cache = client.cache(TABLE)) {
>             for (int i = 0; i < 20; i++)
>                 cache.put(i, Integer.toString(i));
>             ContinuousQuery<Integer, String> query = new ContinuousQuery<>();
>             query.setRemoteFilterFactory(() -> IgniteContinuousQueryExample::isInteresting);
>             query.setInitialQuery(new ScanQuery<>(IgniteContinuousQueryExample::isInteresting));
>             query.setLocalListener(IgniteContinuousQueryExample::handleExistingEntries);
>             try (QueryCursor<Cache.Entry<Integer, String>> resultSet = cache.query(query))
{
>                 handleExistingEntries(resultSet);
>                 // Local listener callbacks will no longer be received once resultSet
is closed so
>                 //  these updates have to be inside the try-with-resources block and
we have to wait
>                 //  to close resultSet after the final update.
>                 for (int i = 20; i < 30; i++)
>                     cache.put(i, Integer.toString(i));
>                 Thread.sleep(60000);
>             }
>         }
>     }
> }
> {code}
> However, this becomes more inconvenient when we want to use SqlQuery in the initial query
to take advantage of indexing. This is the best that I can do:
> query.setRemoteFilterFactory(() -> entry -> entry.getKey() > 10);
>  query.setInitialQuery(new SqlQuery<>(String.sql, "`_key` > 10"));
> This is obviously not ideal because we have to specify the predicate in two different
ways. A quick Google revealed that there are products out there that more seamlessly support
this use case of continuous querying. I understand that Ignite isn't built on top of SQL,
unlike the commercial RDBMSes I found, so maybe this is an out-of-scope feature.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Mime
View raw message