Reliably duplicate SQL cache

classic Classic list List threaded Threaded
4 messages Options
Courtney Robinson Courtney Robinson
Reply | Threaded
Open this post in threaded view
|

Reliably duplicate SQL cache

I've written a work around to use the streamer interface with a ScanQuery to duplicate a cache.
Both are created from SQL using something like this:
repo.query("create table page1(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page1\"")
repo.query("create table page2(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page2\"")
The data is copied, printing the size shows 100 as expected in the test but a SQL query on page2 table returns 0 rows.
def copied = repo.query("SELECT * FROM page2 LIMIT 101")
Gets nothing. The copy function used is below. I'm presuming I've missed a step and the SQL index or something else is not being done. How should this be written to duplicate all data from page1 into page2 table/cache.
public void copy(String fromTableName, String toTableName) {
var ignite = ctx.ignite;
try (
IgniteCache<Object, Object> from = ignite.cache(fromTableName);
IgniteCache<Object, Object> to = ignite.cache(toTableName)
) {
if (from == null || to == null) {
throw new IllegalArgumentException(format("Both from and to tables must exist. from: %s, to: %s", fromTableName, toTableName));
}
try (
IgniteDataStreamer<Object, Object> strmr = ignite.dataStreamer(toTableName/*from.getName()*/);
var cursor = from.withKeepBinary().query(new ScanQuery<>())
) {
strmr.allowOverwrite(true);
strmr.keepBinary(true);
//strmr.receiver(StreamVisitor.from((cache, e) -> to.put(e.getKey(), e.getValue())));
for (Cache.Entry<Object, Object> e : cursor) {
strmr.addData(e.getKey(), e.getValue());
}
//strmr.flush();
}
log.info("Total in target cache {}", to.sizeLong(CachePeekMode.ALL));
}
}


Regards,
Courtney Robinson
Founder and CEO, Hypi

https://hypi.io
ilya.kasnacheev ilya.kasnacheev
Reply | Threaded
Open this post in threaded view
|

Re: Reliably duplicate SQL cache

Hello!

Two tables have different name of an indexed binary type by default.

Try
repo.query("create table page1(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page1, key_type=PageKey, value_type=Page\"")
repo.query("create table page2(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page2, key_type=PageKey, value_type=Page\"")

Regards,
--
Ilya Kasnacheev


сб, 13 февр. 2021 г. в 19:32, Courtney Robinson <[hidden email]>:
I've written a work around to use the streamer interface with a ScanQuery to duplicate a cache.
Both are created from SQL using something like this:
repo.query("create table page1(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page1\"")
repo.query("create table page2(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page2\"")
The data is copied, printing the size shows 100 as expected in the test but a SQL query on page2 table returns 0 rows.
def copied = repo.query("SELECT * FROM page2 LIMIT 101")
Gets nothing. The copy function used is below. I'm presuming I've missed a step and the SQL index or something else is not being done. How should this be written to duplicate all data from page1 into page2 table/cache.
public void copy(String fromTableName, String toTableName) {
var ignite = ctx.ignite;
try (
IgniteCache<Object, Object> from = ignite.cache(fromTableName);
IgniteCache<Object, Object> to = ignite.cache(toTableName)
) {
if (from == null || to == null) {
throw new IllegalArgumentException(format("Both from and to tables must exist. from: %s, to: %s", fromTableName, toTableName));
}
try (
IgniteDataStreamer<Object, Object> strmr = ignite.dataStreamer(toTableName/*from.getName()*/);
var cursor = from.withKeepBinary().query(new ScanQuery<>())
) {
strmr.allowOverwrite(true);
strmr.keepBinary(true);
//strmr.receiver(StreamVisitor.from((cache, e) -> to.put(e.getKey(), e.getValue())));
for (Cache.Entry<Object, Object> e : cursor) {
strmr.addData(e.getKey(), e.getValue());
}
//strmr.flush();
}
log.info("Total in target cache {}", to.sizeLong(CachePeekMode.ALL));
}
}


Regards,
Courtney Robinson
Founder and CEO, Hypi

https://hypi.io
Courtney Robinson Courtney Robinson
Reply | Threaded
Open this post in threaded view
|

Re: Reliably duplicate SQL cache

Hi Illya,
Thanks for responding.
That makes sense - I figured something like that but didn't know exactly what.
Is it possible to get the existing key_type and value_type for tables?
The reason is because we have tables in production and they were not created with key_type and value_type. We actually thought this only applied when you use Java classes with annotations.

In the SYS table somewhere perhaps?


On Mon, Feb 15, 2021 at 11:19 AM Ilya Kasnacheev <[hidden email]> wrote:
Hello!

Two tables have different name of an indexed binary type by default.

Try
repo.query("create table page1(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page1, key_type=PageKey, value_type=Page\"")
repo.query("create table page2(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page2, key_type=PageKey, value_type=Page\"")

Regards,
--
Ilya Kasnacheev


сб, 13 февр. 2021 г. в 19:32, Courtney Robinson <[hidden email]>:
I've written a work around to use the streamer interface with a ScanQuery to duplicate a cache.
Both are created from SQL using something like this:
repo.query("create table page1(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page1\"")
repo.query("create table page2(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page2\"")
The data is copied, printing the size shows 100 as expected in the test but a SQL query on page2 table returns 0 rows.
def copied = repo.query("SELECT * FROM page2 LIMIT 101")
Gets nothing. The copy function used is below. I'm presuming I've missed a step and the SQL index or something else is not being done. How should this be written to duplicate all data from page1 into page2 table/cache.
public void copy(String fromTableName, String toTableName) {
var ignite = ctx.ignite;
try (
IgniteCache<Object, Object> from = ignite.cache(fromTableName);
IgniteCache<Object, Object> to = ignite.cache(toTableName)
) {
if (from == null || to == null) {
throw new IllegalArgumentException(format("Both from and to tables must exist. from: %s, to: %s", fromTableName, toTableName));
}
try (
IgniteDataStreamer<Object, Object> strmr = ignite.dataStreamer(toTableName/*from.getName()*/);
var cursor = from.withKeepBinary().query(new ScanQuery<>())
) {
strmr.allowOverwrite(true);
strmr.keepBinary(true);
//strmr.receiver(StreamVisitor.from((cache, e) -> to.put(e.getKey(), e.getValue())));
for (Cache.Entry<Object, Object> e : cursor) {
strmr.addData(e.getKey(), e.getValue());
}
//strmr.flush();
}
log.info("Total in target cache {}", to.sizeLong(CachePeekMode.ALL));
}
}


Regards,
Courtney Robinson
Founder and CEO, Hypi

https://hypi.io
ilya.kasnacheev ilya.kasnacheev
Reply | Threaded
Open this post in threaded view
|

Re: Reliably duplicate SQL cache

Hello!

It will be available in cache's Query Entities so you can try:
((CacheConfiguration<Object, Object>)ignite.cache("SQL_PUBLIC_<tablename>")
    .getConfiguration(CacheConfiguration.class)).getQueryEntities().iterator().next().getValueType()

Regards,
--
Ilya Kasnacheev


чт, 18 февр. 2021 г. в 11:40, Courtney Robinson <[hidden email]>:
Hi Illya,
Thanks for responding.
That makes sense - I figured something like that but didn't know exactly what.
Is it possible to get the existing key_type and value_type for tables?
The reason is because we have tables in production and they were not created with key_type and value_type. We actually thought this only applied when you use Java classes with annotations.

In the SYS table somewhere perhaps?


On Mon, Feb 15, 2021 at 11:19 AM Ilya Kasnacheev <[hidden email]> wrote:
Hello!

Two tables have different name of an indexed binary type by default.

Try
repo.query("create table page1(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page1, key_type=PageKey, value_type=Page\"")
repo.query("create table page2(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page2, key_type=PageKey, value_type=Page\"")

Regards,
--
Ilya Kasnacheev


сб, 13 февр. 2021 г. в 19:32, Courtney Robinson <[hidden email]>:
I've written a work around to use the streamer interface with a ScanQuery to duplicate a cache.
Both are created from SQL using something like this:
repo.query("create table page1(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page1\"")
repo.query("create table page2(a varchar, b varchar, c varchar, PRIMARY KEY (a, b)) WITH \"cache_name=page2\"")
The data is copied, printing the size shows 100 as expected in the test but a SQL query on page2 table returns 0 rows.
def copied = repo.query("SELECT * FROM page2 LIMIT 101")
Gets nothing. The copy function used is below. I'm presuming I've missed a step and the SQL index or something else is not being done. How should this be written to duplicate all data from page1 into page2 table/cache.
public void copy(String fromTableName, String toTableName) {
var ignite = ctx.ignite;
try (
IgniteCache<Object, Object> from = ignite.cache(fromTableName);
IgniteCache<Object, Object> to = ignite.cache(toTableName)
) {
if (from == null || to == null) {
throw new IllegalArgumentException(format("Both from and to tables must exist. from: %s, to: %s", fromTableName, toTableName));
}
try (
IgniteDataStreamer<Object, Object> strmr = ignite.dataStreamer(toTableName/*from.getName()*/);
var cursor = from.withKeepBinary().query(new ScanQuery<>())
) {
strmr.allowOverwrite(true);
strmr.keepBinary(true);
//strmr.receiver(StreamVisitor.from((cache, e) -> to.put(e.getKey(), e.getValue())));
for (Cache.Entry<Object, Object> e : cursor) {
strmr.addData(e.getKey(), e.getValue());
}
//strmr.flush();
}
log.info("Total in target cache {}", to.sizeLong(CachePeekMode.ALL));
}
}


Regards,
Courtney Robinson
Founder and CEO, Hypi

https://hypi.io