@Test public void testConcurrentIndexedTable() throws InterruptedException, TimeoutException, ExecutionException { QueryContext queryContext = QueryContextConverterUtils .getQueryContextFromSQL("SELECT SUM(m1), MAX(m2) FROM testTable GROUP BY d1, d2, d3 ORDER BY SUM(m1)"); DataSchema dataSchema = new DataSchema(new String[]{"d1", "d2", "d3", "sum(m1)", "max(m2)"}, new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.INT, ColumnDataType.DOUBLE, ColumnDataType.DOUBLE, ColumnDataType.DOUBLE}); IndexedTable indexedTable = new ConcurrentIndexedTable(dataSchema, queryContext, 5); ExecutorService executorService = Executors.newFixedThreadPool(10); try { Callable<Void> c1 = () -> { indexedTable.upsert(getKey(new Object[]{"a", 1, 10d}), getRecord(new Object[]{"a", 1, 10d, 10d, 100d})); indexedTable.upsert(getKey(new Object[]{"b", 2, 20d}), getRecord(new Object[]{"b", 2, 20d, 10d, 200d})); indexedTable.upsert(getKey(new Object[]{"c", 3, 30d}), getRecord(new Object[]{"c", 3, 30d, 10000d, 300d})); indexedTable.upsert(getKey(new Object[]{"d", 4, 40d}), getRecord(new Object[]{"d", 4, 40d, 10d, 400d})); indexedTable.upsert(getKey(new Object[]{"d", 4, 40d}), getRecord(new Object[]{"d", 4, 40d, 10d, 400d})); indexedTable.upsert(getKey(new Object[]{"e", 5, 50d}), getRecord(new Object[]{"e", 5, 50d, 10d, 500d})); return null; }; Callable<Void> c2 = () -> { indexedTable.upsert(getKey(new Object[]{"a", 1, 10d}), getRecord(new Object[]{"a", 1, 10d, 10d, 100d})); indexedTable.upsert(getKey(new Object[]{"f", 6, 60d}), getRecord(new Object[]{"f", 6, 60d, 20000d, 600d})); indexedTable.upsert(getKey(new Object[]{"g", 7, 70d}), getRecord(new Object[]{"g", 7, 70d, 10d, 700d})); indexedTable.upsert(getKey(new Object[]{"b", 2, 20d}), getRecord(new Object[]{"b", 2, 20d, 10d, 200d})); indexedTable.upsert(getKey(new Object[]{"b", 2, 20d}), getRecord(new Object[]{"b", 2, 20d, 10d, 200d})); indexedTable.upsert(getKey(new Object[]{"h", 8, 80d}), getRecord(new Object[]{"h", 8, 80d, 10d, 800d})); indexedTable.upsert(getKey(new Object[]{"a", 1, 10d}), getRecord(new Object[]{"a", 1, 10d, 10d, 100d})); indexedTable.upsert(getKey(new Object[]{"i", 9, 90d}), getRecord(new Object[]{"i", 9, 90d, 500d, 900d})); return null; }; Callable<Void> c3 = () -> { indexedTable.upsert(getKey(new Object[]{"a", 1, 10d}), getRecord(new Object[]{"a", 1, 10d, 10d, 100d})); indexedTable.upsert(getKey(new Object[]{"j", 10, 100d}), getRecord(new Object[]{"j", 10, 100d, 10d, 1000d})); indexedTable.upsert(getKey(new Object[]{"b", 2, 20d}), getRecord(new Object[]{"b", 2, 20d, 10d, 200d})); indexedTable.upsert(getKey(new Object[]{"k", 11, 110d}), getRecord(new Object[]{"k", 11, 110d, 10d, 1100d})); indexedTable.upsert(getKey(new Object[]{"a", 1, 10d}), getRecord(new Object[]{"a", 1, 10d, 10d, 100d})); indexedTable.upsert(getKey(new Object[]{"l", 12, 120d}), getRecord(new Object[]{"l", 12, 120d, 10d, 1200d})); indexedTable.upsert(getKey(new Object[]{"a", 1, 10d}), getRecord(new Object[]{"a", 1, 10d, 10d, 100d})); indexedTable.upsert(getKey(new Object[]{"b", 2, 20d}), getRecord(new Object[]{"b", 2, 20d, 10d, 200d})); indexedTable.upsert(getKey(new Object[]{"m", 13, 130d}), getRecord(new Object[]{"m", 13, 130d, 10d, 1300d})); indexedTable.upsert(getKey(new Object[]{"n", 14, 140d}), getRecord(new Object[]{"n", 14, 140d, 10d, 1400d})); return null; }; List<Future<Void>> futures = executorService.invokeAll(Arrays.asList(c1, c2, c3)); for (Future future : futures) { future.get(10, TimeUnit.SECONDS); } indexedTable.finish(false); Assert.assertEquals(indexedTable.size(), 5); checkEvicted(indexedTable, "c", "f"); } finally { executorService.shutdown(); } }
@Override public boolean upsert(Record record) { Object[] keyValues = Arrays.copyOf(record.getValues(), _numKeyColumns); return upsert(new Key(keyValues), record); }
IndexedTable extends BaseTable { @Override public boolean upsert(Record record) { Object[] keyValues = Arrays.copyOf(record.getValues(), _numKeyColumns); return upsert(new Key(keyValues), record); } }
IndexedTable extends BaseTable { @Override public boolean upsert(Record record) { Object[] keyValues = Arrays.copyOf(record.getValues(), _numKeyColumns); return upsert(new Key(keyValues), record); } protected  IndexedTable(DataSchema dataSchema, QueryContext queryContext, int capacity); }
IndexedTable extends BaseTable { @Override public boolean upsert(Record record) { Object[] keyValues = Arrays.copyOf(record.getValues(), _numKeyColumns); return upsert(new Key(keyValues), record); } protected  IndexedTable(DataSchema dataSchema, QueryContext queryContext, int capacity); @Override boolean upsert(Record record); }
IndexedTable extends BaseTable { @Override public boolean upsert(Record record) { Object[] keyValues = Arrays.copyOf(record.getValues(), _numKeyColumns); return upsert(new Key(keyValues), record); } protected  IndexedTable(DataSchema dataSchema, QueryContext queryContext, int capacity); @Override boolean upsert(Record record);  }