N1ql query return incorrect result and inconsistent results

Java SDK
,
1

Hi,
I’m using java-client 2.4.3 and couchbase server 4.6.
I’m facing a strange behaviour in couchbase: running the following several times returns differents values:

SELECT T., Product., Account.* FROM MyBucket T
JOIN MyBucket Product ON KEYS T.product_ref
JOIN MyBucket Account ON KEYS T.account_ref
where T._type=‘Transaction’ and Product._type = ‘Product’ and Account._type=‘Account’

I’m using the following code:

	N1qlQuery query = N1qlQuery.simple(queryString);
	final AtomicInteger count=new AtomicInteger(0);
	CountDownLatch countDownLatch=new CountDownLatch(1);
	bucket.async().query(query).flatMap(asyncN1qlQueryResult -> asyncN1qlQueryResult.rows())
	.map(asyncN1qlQueryRow -> asyncN1qlQueryRow.value()).subscribe(new Subscriber<JsonObject>() {

		@Override
		public void onCompleted() {
			System.out.println("complete");
			countDownLatch.countDown();
			
		}

		@Override
		public void onError(Throwable e) {
			e.printStackTrace();
			
		}

		@Override
		public void onNext(JsonObject t) {
			//System.out.println("next.."+t);
			count.incrementAndGet();
			
		}

		
	});
	
	try {
		countDownLatch.await();
		System.out.println("Finished in "+((System.currentTimeMillis()-start)/1000));
		System.out.println("Found "+count.intValue());
	} catch (InterruptedException e1) {
		// TODO Auto-generated catch block
		e1.printStackTrace();
	}

thress sequent call of my codes returns these 3 differents result:

Thread…main
complete
Finished in 395
Found 14301

Thread…main
complete
Finished in 346
Found 14491

Thread…main
complete
Finished in 305
Found 13391

More strange is that if I query count(*) instead of querying json fields,I’m getting 109102 which is the correct number of my documents.

Even if I query only the field _type (which is an indexed field), I got the same result as count (109102 documents).

Can anyone explains this behaviour?

Thanks

2

Run the query in cbq shell or QueryWorkbench and check if you get same results.
Is the data is updated/deleted/inserted by another thread.

3

Thanks vsr1
There is no other threads using the bucket.
I have used QUery Workbench but it hangs running the query .

4

It might taking time.
You can try
SELECT COUNT(1) FROM …

Also try cbq shell, Also post sample document of type=‘Transaction’ for fields product_ref, account_ref

Is it works with 4.6.2

5

@zizou you are using the async API and missed two things:

  • you need to chain in a .timeout() operator on the client side (this is done for you on the sync api)
  • you are not consuming the errors right now of the result

Most likely the result is timing out and because you are not consuming the error you are flying blind. Adding a client side timeout similart to the server side timeout set (which has a default or you can customize it on the query params) allows you to better control timeout behavior.

6

Thanks daschl,

Just didn’t get what do you mean by “not consuming the errors”? I’ supposing that my subscriber.onError method is the one consuming error.

For timeout, I have updated my code to look as the following:

bucket.async().query(query).flatMap(asyncN1qlQueryResult → asyncN1qlQueryResult.rows())
.timeout(60, TimeUnit.SECONDS)
.map(asyncN1qlQueryRow → asyncN1qlQueryRow.value())
.subscribe(new Subscriber() {

but still the same issue (no errors and incorrect result)

Can you please help?

7

@zizou, Have a look at @simonbasle’s response in this post: Async (reactive) N1QL.

The errors are emitted by the AsyncN1qlQueryResult.errors() observable. You need to handle those errors in some fashion and emit an exception so your subscriber’s onError() will be invoked. In some situations you may be satisfied with partial results and can do as you did and just map to the AsyncN1qlQueryResult.rows() observable.

I have used code like this to throw an exception for one of the emitted errors:

bucket.async()
    .query(query)
    .flatMap(asyncN1qlQueryResult ->
        asyncN1qlQueryResult.errors()
            .flatMap(error -> 
                 Observable.<AsyncN1qlQueryRow>error(
                     new RuntimeException("Error while performing query: " 
                         + error)))
            .switchIfEmpty(asyncN1qlQueryResult.rows()))
    .map(asyncN1qlQueryRow -> asyncN1qlQueryRow.value())
    ...
1 Like