Add support for Query controller and timeouts; remove vestigial support for interrupts.

7 files changed, 233 insertions, 4 deletions

diff --git a/src/test/java/com/amazon/carbonado/TestStorables.java b/src/test/java/com/amazon/carbonado/TestStorables.java
index ab78a41..20f640d 100644
--- a/src/test/java/com/amazon/carbonado/TestStorables.java
+++ b/src/test/java/com/amazon/carbonado/TestStorables.java
@@ -233,7 +233,6 @@ public class TestStorables extends TestCase {
         catch (UniqueConstraintException e) {
         }
 
-
         Storage<StorableTestMultiPK> storageMPK =
                 getRepository().storageFor(StorableTestMultiPK.class);
         StorableTestMultiPK smpk = storageMPK.prepare();
@@ -250,7 +249,6 @@ public class TestStorables extends TestCase {
     }
 
     public void test_storableStorableStates() throws Exception {
-
         Storage<StorableTestKeyValue> storageMinimal =
             getRepository().storageFor(StorableTestKeyValue.class);
 
@@ -3454,6 +3452,67 @@ public class TestStorables extends TestCase {
         assertNull(order);
     }
 
+    public void test_countTimeout() throws Exception {
+        Storage<StorableTestMinimal> storage =
+            getRepository().storageFor(StorableTestMinimal.class);
+
+        Transaction txn = getRepository().enterTransaction();
+        try {
+            for (int i=0; i<100000; i++) {
+                StorableTestMinimal min = storage.prepare();
+                min.setId(i);
+                min.insert();
+                if (i % 100 == 0) {
+                    txn.commit();
+                }
+            }
+            txn.commit();
+        } finally {
+            txn.exit();
+        }
+
+        try {
+            // Use filter to bypass optimizations.
+            storage.query("id >= ?").with(0).count(Query.Timeout.millis(1));
+            fail();
+        } catch (FetchInterruptedException e) {
+        }
+    }
+
+    public void test_fetchTimeout() throws Exception {
+        Storage<StorableTestBasic> storage =
+            getRepository().storageFor(StorableTestBasic.class);
+
+        Transaction txn = getRepository().enterTransaction();
+        try {
+            for (int i=0; i<100000; i++) {
+                StorableTestBasic stb = storage.prepare();
+                stb.setId(i);
+                stb.setStringProp("str " + Math.random());
+                stb.setIntProp(i);
+                stb.setLongProp(i);
+                stb.setDoubleProp(i);
+                stb.insert();
+                if (i % 100 == 0) {
+                    txn.commit();
+                }
+            }
+            txn.commit();
+        } finally {
+            txn.exit();
+        }
+
+        try {
+            Cursor<StorableTestBasic> cursor =
+                storage.query().orderBy("stringProp").fetch(Query.Timeout.millis(1));
+            while (cursor.hasNext()) {
+                cursor.next();
+            }
+            fail();
+        } catch (FetchInterruptedException e) {
+        }
+    }
+
     private void assertUninitialized(boolean expected, Storable storable, String... properties) {
         for (String property : properties) {
             assertEquals(expected, storable.isPropertyUninitialized(property));
diff --git a/src/test/java/com/amazon/carbonado/cursor/TestCursors.java b/src/test/java/com/amazon/carbonado/cursor/TestCursors.java
index 7c4e3de..3b8b305 100644
--- a/src/test/java/com/amazon/carbonado/cursor/TestCursors.java
+++ b/src/test/java/com/amazon/carbonado/cursor/TestCursors.java
@@ -28,6 +28,8 @@ import junit.framework.TestSuite;
 

 import com.amazon.carbonado.Cursor;

 import com.amazon.carbonado.FetchException;

+import com.amazon.carbonado.FetchInterruptedException;

+import com.amazon.carbonado.Query;

 

 import com.amazon.carbonado.stored.Dummy;

 import com.amazon.carbonado.stored.StorableTestMinimal;

@@ -446,6 +448,23 @@ public class TestCursors extends TestCase {
         compareElements(diff, 1, 1, 1, 1, 2, 3, 3, 3, 3, 4, 4, 4);

     }

 

+    public void testFetchTimeout() throws Exception {

+        Infinite inf = new Infinite();

+        long start = System.nanoTime();

+        Cursor<Element> cursor = ControllerCursor.apply(inf, Query.Timeout.seconds(2));

+        try {

+            while (cursor.hasNext()) {

+                cursor.next();

+            }

+            fail();

+        } catch (FetchInterruptedException e) {

+            long end = System.nanoTime();

+            assertTrue(inf.mClosed);

+            double duration = (end - start) / 1000000000.0d;

+            assertTrue(1.5 <= duration && duration <= 2.5);

+        }

+    }

+

     private Cursor<Element> createElements(int... ids) {

         Arrays.sort(ids);

         Element[] elements = new Element[ids.length];

@@ -511,4 +530,21 @@ public class TestCursors extends TestCase {
             return 0;

         }

     }

+

+    private static class Infinite extends AbstractCursor<Element> {

+        private int mID;

+        boolean mClosed;

+

+        public boolean hasNext() {

+            return true;

+        }

+

+        public Element next() {

+            return new Element(++mID);

+        }

+

+        public void close() {

+            mClosed = true;

+        }

+    }

 }

diff --git a/src/test/java/com/amazon/carbonado/cursor/TestMergeSortBuffer.java b/src/test/java/com/amazon/carbonado/cursor/TestMergeSortBuffer.java
index ea13665..4c27f1f 100644
--- a/src/test/java/com/amazon/carbonado/cursor/TestMergeSortBuffer.java
+++ b/src/test/java/com/amazon/carbonado/cursor/TestMergeSortBuffer.java
@@ -18,6 +18,8 @@
 

 package com.amazon.carbonado.cursor;

 

+import java.lang.reflect.UndeclaredThrowableException;

+

 import java.util.Comparator;

 import java.util.NoSuchElementException;

 import java.util.Random;

@@ -148,6 +150,15 @@ public class TestMergeSortBuffer extends TestCase {
         testBuffer(null, HUGE_BUFFER_SIZE);

     }

 

+    public void testHugeWithTimeout() throws Exception {

+        try {

+            testBuffer(null, HUGE_BUFFER_SIZE, Query.Timeout.millis(100));

+            fail();

+        } catch (UndeclaredThrowableException e) {

+            assertTrue(e.getCause() instanceof FetchInterruptedException);

+        }

+    }

+

     public void testLobs() throws Exception {

         Comparator<StorableWithLobs> c = BeanComparator.forClass(StorableWithLobs.class)

             .orderBy("-id");

@@ -179,8 +190,18 @@ public class TestMergeSortBuffer extends TestCase {
         buffer.close();

     }

 

-    private void testBuffer(Storage<StorableTestBasic> storage, int size) throws Exception {

-        SortBuffer<StorableTestBasic> buffer = new MergeSortBuffer<StorableTestBasic>(storage);

+    private void testBuffer(Storage<StorableTestBasic> storage, int size)

+        throws Exception

+    {

+        testBuffer(storage, size, null);

+    }

+

+    private void testBuffer(Storage<StorableTestBasic> storage, int size,

+                            Query.Controller controller)

+        throws Exception

+    {

+        SortBuffer<StorableTestBasic> buffer =

+            new MergeSortBuffer<StorableTestBasic>(storage, controller);

         buffer.prepare(mComparator);

 

         if (storage == null) {

diff --git a/src/test/java/com/amazon/carbonado/qe/TestIndexedQueryAnalyzer.java b/src/test/java/com/amazon/carbonado/qe/TestIndexedQueryAnalyzer.java
index 524b532..5861023 100644
--- a/src/test/java/com/amazon/carbonado/qe/TestIndexedQueryAnalyzer.java
+++ b/src/test/java/com/amazon/carbonado/qe/TestIndexedQueryAnalyzer.java
@@ -495,18 +495,36 @@ public class TestIndexedQueryAnalyzer extends TestCase {
             return new ArraySortBuffer<S>();

         }

 

+        public SortBuffer<S> createSortBuffer(Query.Controller controller) {

+            return new ArraySortBuffer<S>();

+        }

+

         public long countAll() {

             throw new UnsupportedOperationException();

         }

 

+        public long countAll(Query.Controller controller) {

+            throw new UnsupportedOperationException();

+        }

+

         public Cursor<S> fetchAll() {

             throw new UnsupportedOperationException();

         }

 

+        public Cursor<S> fetchAll(Query.Controller controller) {

+            throw new UnsupportedOperationException();

+        }

+

         public Cursor<S> fetchOne(StorableIndex<S> index, Object[] identityValues) {

             throw new UnsupportedOperationException();

         }

 

+        public Cursor<S> fetchOne(StorableIndex<S> index, Object[] identityValues,

+                                  Query.Controller controller)

+        {

+            throw new UnsupportedOperationException();

+        }

+

         public Query<?> indexEntryQuery(StorableIndex<S> index) {

             return new EmptyQuery();

         }

@@ -516,6 +534,12 @@ public class TestIndexedQueryAnalyzer extends TestCase {
             throw new UnsupportedOperationException();

         }

 

+        public Cursor<S> fetchFromIndexEntryQuery(StorableIndex<S> index, Query<?> indexEntryQuery,

+                                                  Query.Controller controller)

+        {

+            throw new UnsupportedOperationException();

+        }

+

         public Cursor<S> fetchSubset(StorableIndex<S> index,

                                      Object[] identityValues,

                                      BoundaryType rangeStartBoundary,

@@ -527,5 +551,18 @@ public class TestIndexedQueryAnalyzer extends TestCase {
         {

             throw new UnsupportedOperationException();

         }

+

+        public Cursor<S> fetchSubset(StorableIndex<S> index,

+                                     Object[] identityValues,

+                                     BoundaryType rangeStartBoundary,

+                                     Object rangeStartValue,

+                                     BoundaryType rangeEndBoundary,

+                                     Object rangeEndValue,

+                                     boolean reverseRange,

+                                     boolean reverseOrder,

+                                     Query.Controller controller)

+        {

+            throw new UnsupportedOperationException();

+        }

     }

 }

diff --git a/src/test/java/com/amazon/carbonado/qe/TestIndexedQueryExecutor.java b/src/test/java/com/amazon/carbonado/qe/TestIndexedQueryExecutor.java
index 3b62971..0e59f44 100644
--- a/src/test/java/com/amazon/carbonado/qe/TestIndexedQueryExecutor.java
+++ b/src/test/java/com/amazon/carbonado/qe/TestIndexedQueryExecutor.java
@@ -741,6 +741,12 @@ public class TestIndexedQueryExecutor extends TestCase {
             throw new UnsupportedOperationException();

         }

 

+        public Cursor<S> fetchFromIndexEntryQuery(StorableIndex<S> index, Query<?> indexEntryQuery,

+                                                  Query.Controller controller)

+        {

+            throw new UnsupportedOperationException();

+        }

+

         public Cursor<S> fetchSubset(StorableIndex<S> index,

                                      Object[] identityValues,

                                      BoundaryType rangeStartBoundary,

@@ -761,5 +767,25 @@ public class TestIndexedQueryExecutor extends TestCase {
             Collection<S> empty = Collections.emptyList();

             return new IteratorCursor<S>(empty);

         }

+

+        public Cursor<S> fetchSubset(StorableIndex<S> index,

+                                     Object[] identityValues,

+                                     BoundaryType rangeStartBoundary,

+                                     Object rangeStartValue,

+                                     BoundaryType rangeEndBoundary,

+                                     Object rangeEndValue,

+                                     boolean reverseRange,

+                                     boolean reverseOrder,

+                                     Query.Controller controller)

+        {

+            return fetchSubset(index,

+                               identityValues,

+                               rangeStartBoundary,

+                               rangeStartValue,

+                               rangeEndBoundary,

+                               rangeEndValue,

+                               reverseRange,

+                               reverseOrder);

+        }

     }

 }

diff --git a/src/test/java/com/amazon/carbonado/qe/TestJoinedQueryExecutor.java b/src/test/java/com/amazon/carbonado/qe/TestJoinedQueryExecutor.java
index 4eaad57..b4b97d4 100644
--- a/src/test/java/com/amazon/carbonado/qe/TestJoinedQueryExecutor.java
+++ b/src/test/java/com/amazon/carbonado/qe/TestJoinedQueryExecutor.java
@@ -252,18 +252,36 @@ public class TestJoinedQueryExecutor extends TestQueryExecutor {
             return new ArraySortBuffer<S>();

         }

 

+        public SortBuffer<S> createSortBuffer(Query.Controller controller) {

+            return new ArraySortBuffer<S>();

+        }

+

         public long countAll() {

             throw new UnsupportedOperationException();

         }

 

+        public long countAll(Query.Controller controller) {

+            throw new UnsupportedOperationException();

+        }

+

         public Cursor<S> fetchAll() {

             throw new UnsupportedOperationException();

         }

 

+        public Cursor<S> fetchAll(Query.Controller controller) {

+            throw new UnsupportedOperationException();

+        }

+

         public Cursor<S> fetchOne(StorableIndex<S> index, Object[] identityValues) {

             throw new UnsupportedOperationException();

         }

 

+        public Cursor<S> fetchOne(StorableIndex<S> index, Object[] identityValues,

+                                  Query.Controller controller)

+        {

+            throw new UnsupportedOperationException();

+        }

+

         public Query<?> indexEntryQuery(StorableIndex<S> index) {

             return null;

         }

@@ -273,6 +291,12 @@ public class TestJoinedQueryExecutor extends TestQueryExecutor {
             throw new UnsupportedOperationException();

         }

 

+        public Cursor<S> fetchFromIndexEntryQuery(StorableIndex<S> index, Query<?> indexEntryQuery,

+                                                  Query.Controller controller)

+        {

+            throw new UnsupportedOperationException();

+        }

+

         public Cursor<S> fetchSubset(StorableIndex<S> index,

                                      Object[] identityValues,

                                      BoundaryType rangeStartBoundary,

@@ -284,6 +308,19 @@ public class TestJoinedQueryExecutor extends TestQueryExecutor {
         {

             throw new UnsupportedOperationException();

         }

+

+        public Cursor<S> fetchSubset(StorableIndex<S> index,

+                                     Object[] identityValues,

+                                     BoundaryType rangeStartBoundary,

+                                     Object rangeStartValue,

+                                     BoundaryType rangeEndBoundary,

+                                     Object rangeEndValue,

+                                     boolean reverseRange,

+                                     boolean reverseOrder,

+                                     Query.Controller controller)

+        {

+            throw new UnsupportedOperationException();

+        }

     }

 

     static class ScanQuerySupport<S extends Storable> implements FullScanQueryExecutor.Support<S> {

@@ -301,8 +338,16 @@ public class TestJoinedQueryExecutor extends TestQueryExecutor {
             return mQuery.count();

         }

 

+        public long countAll(Query.Controller controller) throws FetchException {

+            return mQuery.count(controller);

+        }

+

         public Cursor<S> fetchAll() throws FetchException {

             return mQuery.fetch();

         }

+

+        public Cursor<S> fetchAll(Query.Controller controller) throws FetchException {

+            return mQuery.fetch(controller);

+        }

     }

 }

diff --git a/src/test/java/com/amazon/carbonado/repo/jdbc/TestH2.java b/src/test/java/com/amazon/carbonado/repo/jdbc/TestH2.java
index 4e4b622..b5e7103 100644
--- a/src/test/java/com/amazon/carbonado/repo/jdbc/TestH2.java
+++ b/src/test/java/com/amazon/carbonado/repo/jdbc/TestH2.java
@@ -383,6 +383,11 @@ public class TestH2 extends com.amazon.carbonado.TestStorables {
     }

 

     @Override

+    public void test_countTimeout() throws Exception {

+        // Timeout granularity is too high. (one second)

+    }

+

+    @Override

     protected BigInteger expected(BigInteger bi) {

         // Used to detect that BigIntegerAdapter was selected.

         return bi.add(BigInteger.ONE);