Source code

001package com.hammurapi.store;
002
003import java.io.Serializable;
004import java.util.ArrayList;
005import java.util.Collection;
006import java.util.Collections;
007import java.util.Comparator;
008import java.util.Iterator;
009import java.util.LinkedList;
010import java.util.List;
011import java.util.Map;
012import java.util.concurrent.Callable;
013import java.util.concurrent.ExecutorService;
014import java.util.concurrent.Future;
015import java.util.concurrent.locks.ReadWriteLock;
016
017import com.hammurapi.common.Util;
018import com.hammurapi.extract.CommutativeOr;
019import com.hammurapi.extract.ComparisonResult;
020import com.hammurapi.extract.ComparisonResult.Type;
021import com.hammurapi.extract.CompositePredicate;
022import com.hammurapi.extract.Extractor;
023import com.hammurapi.extract.IndexedExtractor;
024import com.hammurapi.extract.Predicate;
025import com.hammurapi.extract.True;
026
027/**
028 * Abstract base class for object stores.
029 * @author Pavel Vlasov
030 *
031 * @param <T>
032 * @param <S> Self-type - sub-type of store. Generic parameter for predicates, extractors, update tasks and views. 
033 * S should be an interface for master store, deputy store and views to implement.
034 */
035public abstract class AbstractStore<T, PK, S extends Store<T,PK,S>> implements Store<T, PK, S> {
036        
037        /**
038         * @return Store for the mode with primary key.
039         */
040        protected abstract Map<PK, Handle<T, PK, S>> getPkStore();
041        
042        /**
043         * Store for the mode without primary key or pkStore values collection.
044         */
045        protected abstract Collection<Handle<T, PK, S>> getNoPkStore();
046        
047        /**
048         * @return Executor service to parallel processing.
049         */
050        protected abstract ExecutorService getExecutorService();
051        
052        /**
053         * Creates handle.
054         */
055        protected abstract Handle<T, PK, S> createHandle(T obj, PK primaryKey, Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache, Predicate<T, S>[] validators);
056        
057        /**
058         * Creates collection for index entries
059         * @param unique Collection should be unique (set)
060         * @param ordered Collection should be ordered by the comparator or by natural order.
061         * @param comparator
062         * @return
063         */
064        protected abstract Collection<Handle<T, PK, S>> createIndexCollection(boolean unique, boolean ordered, Comparator<T> comparator);
065        
066        protected Handle<T, PK, S> createHandle(T obj, Predicate<T, S>[] validators) {
067                return createHandle(obj, null, createCache(), validators);
068        }
069        
070        /**
071         * @return Cache for extractors.
072         */
073        protected abstract Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> createCache();
074                
075        /**
076         * Creates a store which encapsulates current call state (e.g. locks currently held) and can be
077         * passed to executor for asynchronous execution. Delegates operate on behalf of the master
078         * store. They don't acquire locks if the master store already holds them.
079         * If the master store holds read lock then attempt of a deputy to acquire write lock will result in
080         * StoreException. 
081         * @return
082         */
083        protected abstract S createDeputy();
084        protected abstract ReadWriteLock createMasterLock();
085        
086        /**
087         * Invoked when some handles have been removed.
088         * This operation shall remove invalidated handles from store and index collections.
089         * This method is invoked outside of the write lock.
090         * @param removed Number of removed handles.
091         */
092        protected abstract void onRemoved(int removed);         
093        
094        /*
095         * Implementation note - removal of objects is done by marking Handle as removed. Handles may or may not be removed from store and index collections
096         * upon marking them as removed. Therefore there should be a "garbage collection" mechanism for removal of invalid handles. 
097         */
098        
099        @SuppressWarnings("unchecked")
100        @Override
101        public Handle<T, PK, S> put(T obj, Predicate<T, S>... validators) {
102                // Evaluate validators.
103                for (Predicate<T, S> v: validators) {
104                        if (!v.extract((S) this, null, Util.wrap(obj))) {
105                                return null;
106                        }
107                }
108                writeLock().lock();
109                try {
110                        if (getPrimaryKeyExtractor()==null) {
111                                Handle<T, PK, S> ret = createHandle(obj, validators);
112                                if (getNoPkStore().add(ret)) {
113                                        updateIndices(ret, true);
114                                } else {
115                                        // Not unique
116                                        for (Handle<T, PK, S> hi: getNoPkStore()) {
117                                                if (ret.equals(hi)) {
118                                                        return hi;
119                                                }
120                                        }
121                                } 
122                                return ret;
123                        } else {
124                                Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache = createCache();                   
125                                PK primaryKey = getPrimaryKeyExtractor().extract((S) this, cache, Util.wrap(obj));
126                                Handle<T, PK, S> hi = getPkStore().get(primaryKey);
127                                if (hi==null) {
128                                        hi = createHandle(obj, primaryKey, cache, validators);
129                                        updateIndices(hi, true);
130                                        getPkStore().put(primaryKey, hi);
131                                } else {
132                                        hi.update(obj);
133                                }
134                                return hi;
135                        }
136                } finally {
137                        writeLock().unlock();
138                }
139        }
140
141        @Override
142        public Iterable<T> getAll() {
143                readLock().lock();
144                try {
145                        Collection<T> ret = new ArrayList<T>();
146                        for (Handle<T, PK, S> hi: getNoPkStore()) {
147                                if (hi.isValid()) {
148                                        ret.add(hi.get());
149                                }
150                        }
151                        return ret;
152                } finally {
153                        readLock().unlock();
154                }
155        }
156
157        @Override
158        public T getByPrimaryKey(PK primaryKey) {
159                readLock().lock();
160                try {
161                        if (getPrimaryKeyExtractor()==null) {
162                                throw new IllegalStateException("Object store is configured without primary key");
163                        }
164                        Handle<T, PK, S> hi = getPkStore().get(primaryKey);
165                        if (hi==null || !hi.isValid()) {
166                                // TODO - get from the backing store, update indices
167                                return null;
168                        }
169                        return hi.get();
170                } finally {
171                        readLock().unlock();
172                }
173        }
174
175        @Override
176        public Iterable<T> get(Predicate<T, S> selector) {
177                Collection<T> ret = new ArrayList<T>();
178                for (Handle<T, PK, S> h: getHandles(selector)) {
179                        ret.add(h.get());
180                }
181                return ret;
182        }
183        
184        protected class SelectorTask implements Callable<Handle<T, PK, S>>, Serializable {
185                
186                private Predicate<T, S> selector;
187                private Handle<T, PK, S> handle;
188
189                public SelectorTask(Handle<T, PK, S> handle, Predicate<T, S> selector) {
190                        this.handle = handle;
191                        this.selector = selector;
192                }
193
194                /**
195                 * Returns object if predicate matches.
196                 */
197                @Override
198                public Handle<T, PK, S> call() throws Exception {
199                        if (handle.isValid() && (selector==null || handle.extract(selector))) {
200                                return handle;
201                        }
202                        return null;
203                }
204                
205        }
206        
207        /**
208         * Result of applying index optimizations to data manipulation parameters.
209         * @author Pavel Vlasov
210         *
211         */
212        private class IndexizationResult {
213                private Iterable<Handle<T, PK, S>> source;
214                private Predicate<T, S> selector;
215                private boolean ordered;
216                
217                public IndexizationResult(
218                                Iterable<Handle<T, PK, S>> source,
219                                Predicate<T, S> selector, 
220                                boolean ordered) {
221                        super();
222                        this.source = source;
223                        this.selector = selector;
224                        this.ordered = ordered;
225                }
226                
227                /**
228                 * @return Data to iterate over for further refinement.
229                 */
230                public Iterable<Handle<T, PK, S>> getSource() {
231                        return source;
232                }
233                
234                /**
235                 * @return Refined selector with parts addressed by indices already removed.
236                 */
237                public Predicate<T, S> getSelector() {
238                        return selector;
239                }
240                
241                /**
242                 * @return true if source is already ordered as required.
243                 */
244                public boolean isOrdered() {
245                        return ordered;
246                }
247                
248        }
249        
250        private <ST extends T> ComparisonResult.Type comparePredicates(
251                        Predicate<T, S> selector,
252                        AbstractIndex<T, ST,PK,?,S> index
253                        ) {
254                
255                if (selector==null) {
256                        if (index.getPredicate()==null || True.getInstance().equals(index.getPredicate())) {
257                                return Type.LESS_RESTRICTIVE;
258                        }
259                        return Type.NOT_EQUAL;
260                }
261                
262                if (index.getPredicate()==null || True.getInstance().equals(index.getPredicate())) {
263                        return Type.LESS_RESTRICTIVE;
264                }
265                
266                ComparisonResult cr = index.getPredicate().compareTo(selector);
267                if (!cr.isOneToOneMapping()) {
268                        return Type.NOT_EQUAL;
269                }
270                
271                switch (cr.getType()) {
272                case EQUAL:
273                case LESS_RESTRICTIVE:
274                        return cr.getType();
275                default:
276                        return Type.NOT_EQUAL;
277                }
278        }
279        
280        /**
281         * @param <V>
282         * @param extractor
283         * @param comparator
284         * @return
285         */
286        private <V, ST extends T> boolean compatibleOrder(Extractor<T, V, S> extractor, Comparator<V> comparator, AbstractIndex<T, ST, PK,V,S> index) {
287                if (index.isOrdered()) {
288                        if ((extractor==null && index.getExtractor()==null) || extractor.equals(index.getExtractor())) {
289                                if ((comparator==null && index.getComparator()==null) || comparator.equals(index.getComparator())) {
290                                        return true;
291                                }
292                        }
293                }
294                return false;
295        }
296                        
297        /**
298         * Apply indexing optimizations.
299         * @return
300         */
301        private <V> IndexizationResult indexize(                  
302                        Predicate<T, S> selector,
303                        final Extractor<T, V, S> extractor, 
304                        final boolean ordered,
305                        Comparator<V> comparator) {
306                
307                class Candidate implements Comparable<Candidate> {
308                        AbstractIndex<T, ? extends T,PK,?,S> index;
309                        ComparisonResult.Type type;
310                        private boolean offersOrdering;
311                        
312                        public Candidate(AbstractIndex<T, ? extends T, PK, ?, S> index, Type type, boolean offersOrdering) {
313                                super();
314                                this.index = index;
315                                this.type = type;
316                                this.offersOrdering = offersOrdering;
317                        }
318
319                        @Override
320                        public int compareTo(Candidate o) {
321                                
322                                // If ordered, indices with compatible order come first, if not, unordered come first
323                                if (ordered) {
324                                        if (offersOrdering) {
325                                                if (!o.offersOrdering) {
326                                                        return -1;
327                                                }
328                                        } else {
329                                                if (o.offersOrdering) {
330                                                        return 1;
331                                                }
332                                        }
333                                } else {
334                                        if (index.isOrdered()) {
335                                                if (!o.index.isOrdered()) {
336                                                        return 1;
337                                                }
338                                        } else {
339                                                if (o.index.isOrdered()) {
340                                                        return -1;
341                                                }
342                                        }
343                                }
344                                
345                                // Predicates equal to selector come first.
346                                if (!o.type.equals(type)) {
347                                        return Type.EQUAL.equals(type) ? -1 : 1;
348                                }
349                                
350                                // Less restrictive predicates come after more restrictive.
351                                if (index.getPredicate()==null) {
352                                        if (o.index.getPredicate()!=null) {
353                                                return 1;
354                                        }
355                                } else {
356                                        if (o.index.getPredicate()==null) {
357                                                return -1;
358                                        }
359                                        
360                                        ComparisonResult cr = index.getPredicate().compareTo(o.index.getPredicate());
361                                        if (cr.isOneToOneMapping()) {
362                                                switch (cr.getType()) {
363                                                case LESS_RESTRICTIVE:
364                                                        return 1;
365                                                case MORE_RESTRICTIVE:
366                                                        return -1;
367                                                }
368                                        }
369                                }
370                                
371                                if (index.isUnique()) {
372                                        if (!o.index.isUnique()) {
373                                                return -1;
374                                        }
375                                } else {
376                                        if (o.index.isUnique()) {
377                                                return 1;
378                                        }
379                                }
380                                
381                                // More costly index comes first
382                                Extractor<? extends T, ?, S> indexExtractor = index.getExtractor();
383                                Extractor<? extends T, ?, S> otherExtractor = o.index.getExtractor();
384                                double indexCost = indexExtractor==null ? 0 : indexExtractor.getCost();
385                                double otherCost = otherExtractor==null ? 0 : otherExtractor.getCost();
386                                if (indexCost>0 && otherCost>0) {
387                                        if (indexCost*o.index.getGranularity()>otherCost*index.getGranularity()) {
388                                                return -1;
389                                        }
390                                        if (indexCost*o.index.getGranularity()<otherCost*index.getGranularity()) {
391                                                return -1;
392                                        }
393                                }
394                                
395                                int gDelta = o.index.getGranularity() - index.getGranularity();
396                                if (gDelta!=0) {
397                                        return gDelta;
398                                }
399                                
400                                // Apply synchronous first, lazy last
401                                int typeCmp = type.compareTo(o.type);
402                                if (typeCmp!=0) {
403                                        return typeCmp;
404                                }
405                                
406                                return hashCode() - o.hashCode();
407                        }
408                                                
409                }
410                
411                List<Candidate> candidates = new ArrayList<Candidate>();
412                for (AbstractIndex<T, ? extends T, PK,?,S> candidate: getIndices()) {
413                        ComparisonResult.Type type = comparePredicates(selector, candidate);
414                        if (!Type.NOT_EQUAL.equals(type)) {
415                                // Add no-predicate indexes only if they offer ordering.
416                                @SuppressWarnings("unchecked")
417                                boolean offersOrdering = ordered && compatibleOrder(extractor, comparator, (AbstractIndex<T,? extends T,PK,V,S>) candidate);
418                                if (candidate.getPredicate()!=null || offersOrdering) {                         
419                                        candidates.add(new Candidate(candidate, type, offersOrdering));
420                                }
421                        }
422                }
423                
424                if (!candidates.isEmpty()) {
425                        Collections.sort(candidates);
426                        Candidate indexToUse = candidates.get(0);
427                        if (Type.EQUAL.equals(indexToUse.type)) {
428                                return new IndexizationResult(indexToUse.index.getHandles(), null, indexToUse.offersOrdering);                          
429                        } else if (selector instanceof CompositePredicate) { 
430                                @SuppressWarnings("unchecked")
431                                Predicate<T, S> newSelector = ((CompositePredicate<T, ?, S, ?>) selector).remove(indexToUse.index.getPredicate());
432                                return new IndexizationResult(indexToUse.index.getHandles(), newSelector, indexToUse.offersOrdering);
433                        }
434                        // Less restrictive, not composite
435                        return new IndexizationResult(indexToUse.index.getHandles(), selector, indexToUse.offersOrdering);                              
436                }
437                                
438                /**
439                 * Or selector - we can use multiple indices to select, the we have to sort anyway.
440                 */
441                if (selector instanceof CommutativeOr && ((CommutativeOr) selector).getParts().size()>1) {
442                        Collection<Future<IndexizationResult>> indexizationResults = new LinkedList<Future<IndexizationResult>>();
443                        for (Predicate<T, S> part: ((CommutativeOr<T,S>) selector).getParts()) {
444                                indexizationResults.add(getExecutorService().submit(new IndexizationTask<V>(part, extractor)));
445                        }
446                        
447                        Collection<Handle<T, PK, S>> ret = new LinkedList<Handle<T, PK, S>>();
448                        Collection<Future<Handle<T, PK, S>>> fret = new LinkedList<Future<Handle<T, PK, S>>>();
449                        
450                        for (Future<IndexizationResult> fir: indexizationResults) {
451                                try {
452                                        IndexizationResult ir = fir.get();
453                                        if (ir.getSelector()==null || True.getInstance().equals(ir.getSelector())) {
454                                                for (Handle<T, PK, S> h: ir.getSource()) {
455                                                        ret.add(h);
456                                                }
457                                        } else {
458                                                for (Handle<T, PK, S> h: ir.getSource()) {
459                                                        fret.add(getExecutorService().submit(new SelectorTask(h, ir.getSelector())));
460                                                }
461                                        }
462                                } catch (Exception e) {
463                                        throw new StoreException(e);
464                                }
465                        }
466                        
467                        for (Future<Handle<T, PK, S>> fh: fret) {
468                                try {
469                                        Handle<T, PK, S> h = fh.get();
470                                        if (h!=null) {
471                                                ret.add(h);
472                                        }
473                                } catch (Exception e) {
474                                        throw new StoreException(e);
475                                }
476                        }
477                        
478                        return new IndexizationResult(ret, null, false);
479                }               
480                
481                
482                return new IndexizationResult(getNoPkStore(), selector, false);         
483        }
484        
485        protected class IndexizationTask<V> implements Callable<IndexizationResult> {
486
487                private Predicate<T, S> selector;
488                private Extractor<T, V, S> extractor;
489
490                public IndexizationTask(Predicate<T, S> selector, Extractor<T, V, S> extractor) {
491                        this.selector = selector;
492                        this.extractor = extractor;
493                }
494                
495                @Override
496                public IndexizationResult call() throws Exception {
497                        return indexize(selector, extractor, false, null);
498                }
499                
500        }
501
502        /**
503         * @param selector
504         * @return
505         */
506        protected Iterable<Handle<T, PK, S>> getHandles(Predicate<T, S> selector) {
507                readLock().lock();
508                try {
509                        IndexizationResult ir = indexize(selector, null, false, null);
510                        Collection<Future<Handle<T, PK, S>>> selectResults = new ArrayList<Future<Handle<T, PK, S>>>();
511                        for (Handle<T, PK, S> h: ir.getSource()) {
512                                selectResults.add(getExecutorService().submit(new SelectorTask(h, ir.getSelector())));
513                        }
514                        Collection<Handle<T, PK, S>> ret = new ArrayList<Handle<T, PK, S>>();
515                        for (Future<Handle<T, PK, S>> f: selectResults) {
516                                try {
517                                        Handle<T, PK, S> result = f.get();
518                                        if (result!=null) {
519                                                ret.add(result);
520                                        }
521                                } catch (Exception e) {
522                                        throw new StoreException(e);
523                                }
524                        }
525                        return ret;
526                } finally {
527                        readLock().unlock();
528                }
529        }
530
531        protected class ExtractorTask<V> implements Callable<V>, Serializable {
532                
533                private Predicate<T, S> selector;
534                private Handle<T, PK, S> handle;
535                private Extractor<T, V, S> extractor;
536
537                public ExtractorTask(Handle<T, PK, S> handle, Predicate<T, S> selector, Extractor<T, V, S> extractor) {
538                        this.handle = handle;
539                        this.selector = selector;
540                        this.extractor = extractor;
541                }
542
543                /**
544                 * Returns object if predicate matches.
545                 */
546                @Override
547                public V call() throws Exception {
548                        if (handle.isValid() && (selector==null || handle.extract(selector))) {                         
549                                if (extractor==null) {
550                                        return ((V) handle.get());
551                                }
552                                return handle.extract(extractor);
553                        }
554                        return null;
555                }
556                
557        }
558        
559        @Override
560        public <V> Iterable<V> get(
561                        Predicate<T, S> selector,
562                        Extractor<T, V, S> extractor, 
563                        boolean ordered,
564                        Comparator<V> comparator) {
565                
566                readLock().lock();
567                try {
568                        IndexizationResult ir = indexize(selector, extractor, ordered, comparator);
569                        Collection<Future<V>> extractResults = new ArrayList<Future<V>>();
570                        for (Handle<T, PK, S> h: ir.getSource()) {
571                                extractResults.add(getExecutorService().submit(new ExtractorTask<V>(h, ir.getSelector(), extractor)));
572                        }
573                        List<V> ret = new ArrayList<V>();
574                        for (Future<V> f: extractResults) {
575                                try {
576                                        V result = f.get();
577                                        if (result!=null) {
578                                                ret.add(result);
579                                        }
580                                } catch (Exception e) {
581                                        throw new StoreException(e);
582                                }
583                        }
584                        if (ordered && !ir.isOrdered()) {
585                                if (comparator==null) {
586                                        // Trick.
587                                        Collections.sort(ret, new NaturalComparator<V>());
588                                } else {
589                                        Collections.sort(ret, comparator);
590                                }
591                        }
592                        return ret;
593                } finally {
594                        readLock().unlock();
595                }
596        }
597        
598        protected class ValueSelectorTask<V> implements Callable<V>, Serializable {
599                
600                private Predicate<V, S> selector;
601                private V value;
602
603                public ValueSelectorTask(V value, Predicate<V, S> selector) {
604                        this.value = value;
605                        this.selector = selector;
606                }
607
608                /**
609                 * Returns object if predicate matches.
610                 */
611                @Override
612                public V call() throws Exception {
613                        if (selector.extract(createDeputy(), null, Util.wrap(value))) {
614                                return value;
615                        }
616                        return null;
617                }
618                
619        }               
620
621        @Override
622        public <V> Iterable<V> getMultiple(
623                        Predicate<T, S> selector,
624                        Extractor<T, Iterable<V>, S> extractor, 
625                        Predicate<V, S> valueSelector,                                    
626                        boolean ordered,
627                        Comparator<V> comparator) {
628                readLock().lock();
629                try {
630//                      IndexizationResult ir = indexize(selector, extractor, ordered, comparator);
631                        Collection<Future<Iterable<V>>> extractResults = new ArrayList<Future<Iterable<V>>>();
632                        for (Handle<T, PK, S> h: getNoPkStore()) {
633                                extractResults.add(getExecutorService().submit(new ExtractorTask<Iterable<V>>(h, selector, extractor)));
634                        }
635                        List<V> ret = new ArrayList<V>();
636                        if (valueSelector==null) {
637                                for (Future<Iterable<V>> f: extractResults) {
638                                        try {
639                                                Iterable<V> result = f.get();
640                                                if (result!=null) {
641                                                        for (V v: result) {
642                                                                ret.add(v);
643                                                        }
644                                                }
645                                        } catch (Exception e) {
646                                                throw new StoreException(e);
647                                        }
648                                }
649                        } else {
650                                List<Future<V>> fRet = new LinkedList<Future<V>>();
651                                for (Future<Iterable<V>> f: extractResults) {
652                                        try {
653                                                Iterable<V> result = f.get();
654                                                if (result!=null) {
655                                                        for (V v: result) {
656                                                                fRet.add(getExecutorService().submit(new ValueSelectorTask<V>(v, valueSelector)));
657                                                        }
658                                                }
659                                        } catch (Exception e) {
660                                                throw new StoreException(e);
661                                        }
662                                }
663                                for (Future<V> f: fRet) {
664                                        try {
665                                                V result = f.get();
666                                                if (result!=null) {
667                                                        ret.add(result);
668                                                }
669                                        } catch (Exception e) {
670                                                throw new StoreException(e);
671                                        }
672                                }                               
673                        }
674                        if (ordered) {
675                                if (comparator==null) {
676                                        // Trick.
677                                        Collections.sort(ret, new NaturalComparator<V>());
678                                } else {
679                                        Collections.sort(ret, comparator);
680                                }
681                        }
682                        return ret;
683                } finally {
684                        readLock().unlock();
685                }
686        }
687
688        @Override
689        public void clear() {
690                writeLock().lock();
691                try {
692                        // Marking handles as removed.
693                        for (Handle<T, PK, S> h: getNoPkStore()) {
694                                h.remove();
695                        }
696                        // TODO - clear indices, remove from the backing store.
697                        if (getPrimaryKeyExtractor()==null) {
698                                getNoPkStore().clear();
699                        } else {
700                                getPkStore().clear();
701                        }
702                } finally {
703                        writeLock().unlock();
704                }
705        }
706
707        @Override
708        public boolean remove(T obj) {
709                int cleanedUp = 0;
710                writeLock().lock();
711                try {
712                        Iterator<Handle<T, PK, S>> it = getNoPkStore().iterator();
713                        while (it.hasNext()) {
714                                Handle<T, PK, S> next = it.next();
715                                if (obj.equals(next.get())) {
716                                        next.remove(); // Handles shall just invalidate self without modifying the containing collection.
717                                        it.remove();
718                                        // TODO remove from indices, remove from the backing store.
719                                        return true;
720                                }
721                                // Clean up also directly removed (through handle.remove()) entries.
722                                if (!next.isValid()) {
723                                        it.remove();
724                                        ++cleanedUp;
725                                }
726                        }
727                        return false;
728                } finally {
729                        writeLock().unlock();
730                        onRemoved(-cleanedUp); 
731                }
732        }
733        
734        public boolean removeByPrimaryKey(PK primaryKey) {
735                writeLock().lock();
736                try {
737                        if (getPrimaryKeyExtractor()==null) {
738                                throw new IllegalStateException("Object store is configured without primary key");
739                        }
740                        Handle<T, PK, S> hi = getPkStore().remove(primaryKey);
741                        if (hi==null || !hi.isValid()) {
742                                // TODO - get from the backing store, update indices
743                                return false;
744                        }
745                        hi.remove();
746                        return true;
747                } finally {
748                        writeLock().unlock();
749                }
750                
751        };
752
753        protected class RemoveTask implements Callable<Boolean>, Serializable {
754                
755                private Predicate<T, S> selector;
756                private Handle<T, PK, S> handle;
757
758                public RemoveTask(Handle<T, PK, S> handle, Predicate<T, S> selector) {
759                        this.handle = handle;
760                        this.selector = selector;
761                }
762
763                /**
764                 * Removes object and returns true if predicate matches.
765                 */
766                @Override
767                public Boolean call() throws Exception {
768                        if (handle.isValid() && handle.extract(selector)) {
769                                handle.remove(); // TODO - remove from the backing store, shall be handled by handle.remove().
770                                return true;
771                        }
772                        return false;
773                }
774                
775        }
776        
777        @Override
778        public int remove(Predicate<T, S> selector) {
779                int ret = 0;
780                writeLock().lock();
781                try {
782                        IndexizationResult ir = indexize(selector, null, false, null);
783                        Collection<Future<Boolean>> removeResults = new ArrayList<Future<Boolean>>();
784                        for (Handle<T, PK, S> h: ir.getSource()) {
785                                removeResults.add(getExecutorService().submit(new RemoveTask(h, ir.getSelector())));
786                        }
787                        for (Future<Boolean> f: removeResults) {
788                                try {
789                                        if (f.get()) {
790                                                ++ret;
791                                        }
792                                } catch (Exception e) {
793                                        throw new StoreException(e);
794                                }
795                        }
796                        return ret;
797                } finally {
798                        writeLock().unlock();
799                        onRemoved(ret);
800                }
801        }
802
803        protected class UpdateCallable implements Callable<Boolean>, Serializable {
804                
805                private Predicate<T, S> selector;
806                private Handle<T, PK, S> handle;
807                private UpdateTask<T, PK, S> updater;
808                private S storeDeputy;
809
810                public UpdateCallable(Handle<T, PK, S> handle, Predicate<T, S> selector, UpdateTask<T,PK,S> updater, S storeDeputy) {
811                        this.handle = handle;
812                        this.selector = selector;
813                        this.updater = updater;
814                        this.storeDeputy = storeDeputy;
815                }
816
817                /**
818                 * Updates object and returns true if predicate matches.
819                 */
820                @Override
821                public Boolean call() throws Exception {
822                        if (handle.isValid() && handle.extract(selector)) {
823                                updater.execute(storeDeputy, handle); // TODO - update backing store.
824                                return true;
825                        }
826                        return false;
827                }
828                
829        }
830        
831        @Override
832        public int update(Predicate<T, S> selector, UpdateTask<T, PK, S> updater) {
833                writeLock().lock();
834                try {
835                        IndexizationResult ir = indexize(selector, null, false, null);
836                        Collection<Future<Boolean>> updateResults = new ArrayList<Future<Boolean>>();
837                        for (Handle<T, PK, S> h: ir.getSource()) {
838                                updateResults.add(getExecutorService().submit(new UpdateCallable(h, ir.getSelector(), updater, createDeputy())));
839                        }
840                        int ret = 0;
841                        for (Future<Boolean> f: updateResults) {
842                                try {
843                                        if (f.get()) {
844                                                ++ret;
845                                        }
846                                } catch (Exception e) {
847                                        throw new StoreException(e);
848                                }
849                        }
850                        return ret;
851                } finally {
852                        writeLock().unlock();
853                }
854        }
855        
856        protected abstract Collection<AbstractIndex<T,? extends T,PK, ?, S>> getIndices();
857        
858        protected abstract <V, ST extends T> AbstractIndex<T,ST,PK,V,S> createIndex(
859                        Predicate<T, S> predicate,
860                        Extractor<ST, V, S> extractor,
861                        Index.Type type,
862                        boolean ordered, 
863                        Comparator<V> comparator);
864                
865        @SuppressWarnings("unchecked")
866        @Override
867        public <V, ST extends T> Index<T,ST,PK,V,S> addIndex(
868                        Predicate<T, S> predicate,
869                        Extractor<ST, V, S> extractor,
870                        Index.Type type,
871                        boolean ordered, 
872                        Comparator<V> comparator) {
873                
874                writeLock().lock();
875                try {
876                        for (AbstractIndex<T, ? extends T, PK, ?, S> index: getIndices()) {
877                                if ((predicate==null && index.getPredicate()==null) || (predicate!=null && predicate.equals(index.getPredicate()))) {
878                                        if ((extractor == null && index.getExtractor()==null) || (extractor!=null && extractor.equals(index))) {
879                                                if (ordered) {
880                                                        if (!index.isOrdered()) {
881                                                                // Upgrade index to ordered
882                                                                ((AbstractIndex<T, ST, PK,V,S>) index).upgrade(type, ordered, comparator);
883                                                                return (Index<T, ST, PK, V, S>) index;
884                                                        } else {
885                                                                if ((index.getComparator()==null && comparator==null) || (comparator!=null && comparator.equals(index.getComparator()))) {
886                                                                        if (index.type.compareTo(type)>0) {
887                                                                                ((AbstractIndex<T,ST,PK,V,S>) index).upgrade(type, ordered, comparator);
888                                                                        }
889                                                                        return (Index<T, ST, PK, V, S>) index;                                                                    
890                                                                }
891                                                        }
892                                                } else {
893                                                        // Requested index not ordered, doesn't matter if existing index is ordered.
894                                                        // Upgrade type if required.
895                                                        if (index.type.compareTo(type)>0) {
896                                                                ((AbstractIndex<T,ST,PK,V,S>) index).upgrade(type, ordered, comparator);
897                                                        }
898                                                        return (Index<T, ST, PK, V, S>) index;
899                                                }
900                                        }
901                                }
902                        }
903                        AbstractIndex<T, ST, PK, V, S> ret = createIndex(predicate, extractor, type, ordered, comparator);
904                        for (Handle<T, PK, S> hi: getNoPkStore()) {
905                                if (hi.isValid()) {
906                                        ret.update(hi, true);
907                                }
908                        }
909                        getIndices().add(ret);
910                        return ret.createProxy();
911                } finally {
912                        writeLock().unlock();
913                }
914                
915        }
916        
917        private final Extractor<T,T, S> SELF_EXTRACTOR = new IndexedExtractor<T, S>(0);
918
919        protected class QueryCallable<V> implements Callable<Boolean>, Serializable {
920                
921                private Predicate<T, S> selector;
922                private Handle<T, PK, S> handle;
923                private QueryTask<V, PK, S> query;
924                private S storeDeputy;
925                private Extractor<T, V, S> extractor;
926
927                @SuppressWarnings("unchecked")
928                public QueryCallable(
929                                Handle<T, PK, S> handle, 
930                                Predicate<T, S> selector,
931                                Extractor<T, V, S> extractor,
932                                QueryTask<V,PK, S> query, 
933                                S storeDeputy) {
934                        this.handle = handle;
935                        this.selector = selector;
936                        this.query = query;
937                        this.storeDeputy = storeDeputy.createUnmodifiableFacade();
938                        this.extractor = extractor;
939                        if (this.extractor==null) {
940                                this.extractor = (Extractor<T, V, S>) SELF_EXTRACTOR;
941                        }
942                }
943
944                /**
945                 * Updates object and returns true if predicate matches.
946                 */
947                @Override
948                public Boolean call() throws Exception {
949                        if (handle.isValid() && (selector==null || handle.extract(selector))) {
950                                query.execute(storeDeputy, handle.extract(extractor), handle.getPrimaryKey());
951                                return true;
952                        }
953                        return false;
954                }
955                
956        }
957                
958        @Override
959        public int queryAll(final QueryTask<T, PK, S> query) {
960                readLock().lock();
961                try {
962                        Collection<Future<Boolean>> queryResults = new ArrayList<Future<Boolean>>();
963                        for (Handle<T, PK, S> h: getNoPkStore()) {
964                                queryResults.add(getExecutorService().submit(new QueryCallable<T>(h, null, null, query, createDeputy())));
965                        }
966                        int ret = 0;
967                        for (Future<Boolean> f: queryResults) {
968                                try {
969                                        if (f.get()) {
970                                                ++ret;
971                                        }
972                                } catch (Exception e) {
973                                        throw new StoreException(e);
974                                }
975                        }
976                        return ret;
977                } finally {
978                        readLock().unlock();
979                }
980        }
981
982        @Override
983        public int query(Predicate<T, S> selector, QueryTask<T, PK, S> query) {
984                readLock().lock();
985                try {
986                        IndexizationResult ir = indexize(selector, null, false, null);
987                        Collection<Future<Boolean>> queryResults = new ArrayList<Future<Boolean>>();
988                        for (Handle<T, PK, S> h: ir.getSource()) {
989                                queryResults.add(getExecutorService().submit(new QueryCallable<T>(h, ir.getSelector(), null, query, createDeputy())));
990                        }
991                        int ret = 0;
992                        for (Future<Boolean> f: queryResults) {
993                                try {
994                                        if (f.get()) {
995                                                ++ret;
996                                        }
997                                } catch (Exception e) {
998                                        throw new StoreException(e);
999                                }
1000                        }
1001                        
1002                        return ret;
1003                } finally {
1004                        readLock().unlock();
1005                }
1006        }
1007        
1008        @Override
1009        public <V> int query(
1010                        Predicate<T, S> selector,
1011                        Extractor<T, V, S> extractor,
1012                        QueryTask<V, PK, S> query) {
1013                
1014                readLock().lock();
1015                try {
1016                        IndexizationResult ir = indexize(selector, null, false, null);
1017                        Collection<Future<Boolean>> queryResults = new ArrayList<Future<Boolean>>();
1018                        for (Handle<T, PK, S> h: ir.getSource()) {
1019                                queryResults.add(getExecutorService().submit(new QueryCallable<V>(h, ir.getSelector(), extractor, query, createDeputy())));
1020                        }
1021                        int ret = 0;
1022                        for (Future<Boolean> f: queryResults) {
1023                                try {
1024                                        if (f.get()) {
1025                                                ++ret;
1026                                        }
1027                                } catch (Exception e) {
1028                                        throw new StoreException(e);
1029                                }
1030                        }
1031                        
1032                        return ret;
1033                } finally {
1034                        readLock().unlock();
1035                }
1036        }
1037        
1038        protected class QueryMultipleCallable<V> implements Callable<Collection<Future<Integer>>>, Serializable {
1039                
1040                private Predicate<T, S> selector;
1041                private Handle<T, PK, S> handle;
1042                private QueryTask<V, PK, S> query;
1043                private S storeDeputy;
1044                private Extractor<T, Iterable<V>, S> extractor;
1045                private Predicate<V, S> valueSelector;
1046
1047                public QueryMultipleCallable(
1048                                Handle<T, PK, S> handle, 
1049                                Predicate<T, S> selector,
1050                                Extractor<T, Iterable<V>, S> extractor,
1051                                Predicate<V, S> valueSelector,                                            
1052                                QueryTask<V,PK,S> query, 
1053                                S storeDeputy) {
1054                        this.handle = handle;
1055                        this.selector = selector;
1056                        this.query = query;
1057                        this.storeDeputy = storeDeputy;
1058                        this.extractor = extractor;
1059                        this.valueSelector = valueSelector;
1060                }
1061
1062                /**
1063                 * Updates object and returns true if predicate matches.
1064                 */
1065                @Override
1066                public Collection<Future<Integer>> call() throws Exception {
1067                        if (handle.isValid() && (selector==null || handle.extract(selector))) {
1068                                Collection<Future<Integer>> queryItemTasks = new LinkedList<Future<Integer>>();
1069                                for (V v: handle.extract(extractor)) {
1070                                        queryItemTasks.add(getExecutorService().submit(new QueryItemCallable<V>(v, valueSelector, handle.getPrimaryKey(), query, storeDeputy)));
1071                                }
1072                                return queryItemTasks;
1073                        }
1074                        return Collections.emptyList();
1075                }
1076                
1077        }
1078        
1079        protected class QueryItemCallable<V> implements Callable<Integer> {
1080
1081                private V v;
1082                private PK primaryKey;
1083                private QueryTask<V, PK, S> query;
1084                private S storeDeputy;
1085                private Predicate<V, S> valueSelector;
1086
1087                public QueryItemCallable(
1088                                V v, 
1089                                Predicate<V, S> valueSelector,
1090                                PK primaryKey, QueryTask<V, PK, S> query, 
1091                                S storeDeputy) {
1092                        this.v = v;
1093                        this.valueSelector = valueSelector;
1094                        this.primaryKey = primaryKey;
1095                        this.query = query;
1096                        this.storeDeputy = storeDeputy.createUnmodifiableFacade();
1097                }
1098
1099                @Override
1100                public Integer call() throws Exception {
1101                        if (valueSelector==null || valueSelector.extract(storeDeputy, null, Util.wrap(v))) {
1102                                query.execute(storeDeputy, v, primaryKey);
1103                        }
1104                        return 1;
1105                }
1106                
1107        }
1108        
1109        @Override
1110        public <V> int queryMultiple(
1111                        Predicate<T, S> selector,
1112                        Extractor<T, Iterable<V>, S> extractor,
1113                        Predicate<V, S> valueSelector,                                    
1114                        QueryTask<V, PK, S> query) {
1115                readLock().lock();
1116                try {
1117                        IndexizationResult ir = indexize(selector, null, false, null);
1118                        Collection<Future<Collection<Future<Integer>>>> queryResults = new ArrayList<Future<Collection<Future<Integer>>>>();
1119                        for (Handle<T, PK, S> h: ir.getSource()) {
1120                                queryResults.add(getExecutorService().submit(new QueryMultipleCallable<V>(h, ir.getSelector(), extractor, valueSelector, query, createDeputy())));
1121                        }
1122                        Collection<Future<Integer>> itemResults = new LinkedList<Future<Integer>>();
1123                        for (Future<Collection<Future<Integer>>> f: queryResults) {
1124                                try {
1125                                        itemResults.addAll(f.get());
1126                                } catch (Exception e) {
1127                                        throw new StoreException(e);
1128                                }
1129                        }
1130                        
1131                        int ret=0;
1132                        for (Future<Integer> f: itemResults) {
1133                                try {
1134                                        ret+=f.get();
1135                                } catch (Exception e) {
1136                                        throw new StoreException(e);
1137                                }
1138                        }
1139                        return ret;
1140                } finally {
1141                        readLock().unlock();
1142                }
1143        }
1144        
1145        /**
1146         * This method is invoked by handle implementations. 
1147         * It shall not be invoked by client code.
1148         * @param handle
1149         * @param isNew
1150         */
1151        protected void updateIndices(Handle<T, PK, S> handle, boolean isNew) {
1152                Iterator<AbstractIndex<T, ? extends T, PK, ?, S>> it = getIndices().iterator();
1153                while (it.hasNext()) {
1154                        ((AbstractIndex<T, ? extends T, PK, ?, S>) it.next()).update(handle, isNew);
1155                }
1156        }
1157        
1158        @Override
1159        public Iterator<T> iterator() {
1160                return getAll().iterator();
1161        }
1162
1163        @Override
1164        public S createView(
1165                        Predicate<T, S> selector, 
1166                        ViewType viewType) {
1167                
1168                switch (viewType) {
1169                case LIVE:
1170                        return createLiveView(selector);
1171                case SYNCHRONOUS:
1172                case ASYNCHRONOUS: 
1173                case LAZY:
1174                default:
1175                        throw new UnsupportedOperationException("View type is not supported: "+viewType);                       
1176                }
1177        }
1178
1179        protected abstract S createLiveView(Predicate<T, S> selector);
1180        
1181//      protected 
1182
1183}