Query SpatialView with Rxjava


#1

Hi everyone,Why use rxjava like that doesn’t print any data? BTW,i use synchronized query can find data.
bucket.async()
.query(query)
.flatMap(AsyncSpatialViewResult::rows)
.flatMap(AsyncSpatialViewRow::document)
.subscribe(row->{
System.out.println(count.get()+" "+row.content());
});


#2

Hi @hubo3085632 in what context do you call this code, and what is your query?

For instance, if this is called from a main() method, since the code will return immediately, it may be that the application exits even before something is returned.


#3

Herer is my query
"ViewQuery query =ViewQuery.from(“test”, “test”).key(“doc_num”).limit(1);"
I exec query in a for loop ,I’m call this query in a main() method,some row.content() was return but very less.


#4

ok so basically, when you use bucket.async() you switch to asynchronous mode. What that means is that the execution won’t block to wait for query results, it will go straight to the line of code following bucket.async().query(...) (in your example).

Side note: This is typically useful in a program architectured around an “Event Loop”, for example desktop Swing applications, where a single thread deals with lots of short-burst UI updates and processing but must never be blocked (otherwise the UI doesn’t refresh nor react to keyboard and mouse input).

But such application generally have a separate thread for the Event Loop and the main thread is kept alive until the user exits the application (for instance by closing the main window).

So here you only have a main. What happens if execution continues immediately after you fired your query? The main() method ends, which means your application is terminated. Sometimes it happens quickly enough that responses from the query haven’t started coming in yet.

For a simple example like this, you have to block and make your main() wait for the last result to come in. Here is how you can do that:

bucket.async()
  .query(query)
  .flatMap(AsyncSpatialViewResult::rows)
  .flatMap(AsyncSpatialViewRow::document)
  //here we output rows as they come in, but using doOnNext
  .doOnNext(row-> { System.out.println(count.get()+" "+row.content()); })
  //here we block until the last element is received. if no row, returns null
  .toBlocking().lastOrDefault(null);

The benefit of this version over the basic bucket one is that you will see (notification style) each row as it arrives, whereas the basic blocking version first waits for all rows to be returned by the server then allow you to iterate on them.

Notes: I’ve used lastOrDefault(null) because last() will expect at least one element and will throw a NoSuchElementException otherwise, but at this point we don’t really care about the result, we’ve already manipulated the rows.
However note that toBlocking() deals with errors propagated through the Observable by throwing them.


#5

I got it,thank you for your detailed answer :smile:
I meet a weird situation,the other day i query spatialview i find it is slow,today i have done rebalance,then i query spatialview again,it is 30 times faster than before :open_mouth: .Does rebalance should always do or the view index should rebuild?


#6

hey @hubo3085632

I talked to @vmx about that. 30x is quite a difference, but there are two distinct algorithms at play (initial building of the index and incremental updates) and the first one scales better.

So it makes sense that queries done on a bucket that gets a large amount of changes (data added, deleted, expired) would perform worse. Was that your case when you were querying the view with slow response times?

Doing a compaction or a rebalance will cause the indexes to be rebuilt from scratch, which can end up in spatial queries performing faster.


#7

thanks a lot,maybe i should compact my view always.Does all view’s map function will read docs’s type when a new doc come in?


#8

I’m not sure I understand your question. You have to code your views map functions to read the appropriate type field in the content (if any)… All the logic is yours to implement, the view will pass every new document that comes in to your map function.