diff options
Diffstat (limited to 'src/main/java/com/amazon/carbonado/repo/indexed')
5 files changed, 97 insertions, 35 deletions
diff --git a/src/main/java/com/amazon/carbonado/repo/indexed/IndexEntryAccessor.java b/src/main/java/com/amazon/carbonado/repo/indexed/IndexEntryAccessor.java index b107a6e..0c1e6b8 100644 --- a/src/main/java/com/amazon/carbonado/repo/indexed/IndexEntryAccessor.java +++ b/src/main/java/com/amazon/carbonado/repo/indexed/IndexEntryAccessor.java @@ -20,7 +20,6 @@ package com.amazon.carbonado.repo.indexed; import java.util.Comparator;
-import com.amazon.carbonado.FetchException;
import com.amazon.carbonado.Storable;
import com.amazon.carbonado.Storage;
diff --git a/src/main/java/com/amazon/carbonado/repo/indexed/IndexEntryGenerator.java b/src/main/java/com/amazon/carbonado/repo/indexed/IndexEntryGenerator.java index 4059ac3..a126224 100644 --- a/src/main/java/com/amazon/carbonado/repo/indexed/IndexEntryGenerator.java +++ b/src/main/java/com/amazon/carbonado/repo/indexed/IndexEntryGenerator.java @@ -24,7 +24,6 @@ import java.util.WeakHashMap; import java.lang.ref.Reference;
import java.lang.ref.SoftReference;
-import com.amazon.carbonado.FetchException;
import com.amazon.carbonado.Storable;
import com.amazon.carbonado.SupportException;
import com.amazon.carbonado.info.StorableIndex;
diff --git a/src/main/java/com/amazon/carbonado/repo/indexed/IndexedRepository.java b/src/main/java/com/amazon/carbonado/repo/indexed/IndexedRepository.java index d558950..097185a 100644 --- a/src/main/java/com/amazon/carbonado/repo/indexed/IndexedRepository.java +++ b/src/main/java/com/amazon/carbonado/repo/indexed/IndexedRepository.java @@ -43,7 +43,7 @@ import com.amazon.carbonado.info.StorableIntrospector; import com.amazon.carbonado.qe.RepositoryAccess;
import com.amazon.carbonado.qe.StorageAccess;
-import com.amazon.carbonado.spi.StorageCollection;
+import com.amazon.carbonado.spi.StoragePool;
/**
* Wraps another repository in order to make it support indexes. The wrapped
@@ -62,20 +62,23 @@ class IndexedRepository implements Repository, private final String mName;
private final boolean mIndexRepairEnabled;
private final double mIndexThrottle;
- private final StorageCollection mStorages;
+ private final boolean mAllClustered;
+ private final StoragePool mStoragePool;
IndexedRepository(AtomicReference<Repository> rootRef, String name,
Repository repository,
boolean indexRepairEnabled,
- double indexThrottle)
+ double indexThrottle,
+ boolean allClustered)
{
mRootRef = rootRef;
mRepository = repository;
mName = name;
mIndexRepairEnabled = indexRepairEnabled;
mIndexThrottle = indexThrottle;
+ mAllClustered = allClustered;
- mStorages = new StorageCollection() {
+ mStoragePool = new StoragePool() {
protected <S extends Storable> Storage<S> createStorage(Class<S> type)
throws RepositoryException
{
@@ -112,7 +115,7 @@ class IndexedRepository implements Repository, public <S extends Storable> Storage<S> storageFor(Class<S> type)
throws MalformedTypeException, SupportException, RepositoryException
{
- return mStorages.storageFor(type);
+ return mStoragePool.get(type);
}
public Transaction enterTransaction() {
@@ -221,4 +224,8 @@ class IndexedRepository implements Repository, double getIndexRepairThrottle() {
return mIndexThrottle;
}
+
+ boolean isAllClustered() {
+ return mAllClustered;
+ }
}
diff --git a/src/main/java/com/amazon/carbonado/repo/indexed/IndexedRepositoryBuilder.java b/src/main/java/com/amazon/carbonado/repo/indexed/IndexedRepositoryBuilder.java index 4cd4afb..e7ad8ed 100644 --- a/src/main/java/com/amazon/carbonado/repo/indexed/IndexedRepositoryBuilder.java +++ b/src/main/java/com/amazon/carbonado/repo/indexed/IndexedRepositoryBuilder.java @@ -49,6 +49,7 @@ public class IndexedRepositoryBuilder extends AbstractRepositoryBuilder { private RepositoryBuilder mRepoBuilder;
private boolean mIndexRepairEnabled = true;
private double mIndexThrottle = 1.0;
+ private boolean mAllClustered;
public IndexedRepositoryBuilder() {
}
@@ -75,7 +76,8 @@ public class IndexedRepositoryBuilder extends AbstractRepositoryBuilder { Repository repo = new IndexedRepository(rootRef, getName(), wrapped,
isIndexRepairEnabled(),
- getIndexRepairThrottle());
+ getIndexRepairThrottle(),
+ isAllClustered());
rootRef.set(repo);
return repo;
}
@@ -167,6 +169,23 @@ public class IndexedRepositoryBuilder extends AbstractRepositoryBuilder { mIndexThrottle = desiredSpeed;
}
+ /**
+ * Returns true if all indexes should be identified as clustered. This
+ * affects how indexes are selected by the query analyzer.
+ */
+ public boolean isAllClustered() {
+ return mAllClustered;
+ }
+
+ /**
+ * When all indexes are identified as clustered, the query analyzer treats
+ * all indexes as performing equally well. This is suitable for indexing
+ * repositories that never read from a slow storage medium.
+ */
+ public void setAllClustered(boolean clustered) {
+ mAllClustered = clustered;
+ }
+
public void errorCheck(Collection<String> messages) throws ConfigurationException {
super.errorCheck(messages);
if (null == getWrappedRepository()) {
diff --git a/src/main/java/com/amazon/carbonado/repo/indexed/IndexedStorage.java b/src/main/java/com/amazon/carbonado/repo/indexed/IndexedStorage.java index 7b66f50..fe0cfe8 100644 --- a/src/main/java/com/amazon/carbonado/repo/indexed/IndexedStorage.java +++ b/src/main/java/com/amazon/carbonado/repo/indexed/IndexedStorage.java @@ -30,6 +30,7 @@ import org.apache.commons.logging.LogFactory; import com.amazon.carbonado.Cursor;
import com.amazon.carbonado.FetchException;
import com.amazon.carbonado.IsolationLevel;
+import com.amazon.carbonado.PersistException;
import com.amazon.carbonado.Query;
import com.amazon.carbonado.RepositoryException;
import com.amazon.carbonado.Storable;
@@ -188,6 +189,9 @@ class IndexedStorage<S extends Storable> implements Storage<S>, StorageAccess<S> // Assume index is malformed, so ignore it.
continue;
}
+ if (mRepository.isAllClustered()) {
+ freeIndex = freeIndex.clustered(true);
+ }
mAllIndexInfoMap.put(freeIndex, ii);
freeIndexSet.add(freeIndex);
}
@@ -210,6 +214,10 @@ class IndexedStorage<S extends Storable> implements Storage<S>, StorageAccess<S> // Add the indexes we get for free.
queryableIndexSet.addAll(freeIndexSet);
+
+ if (mRepository.isAllClustered()) {
+ queryableIndexSet.markClustered(true);
+ }
}
// The set of indexes that should be kept up-to-date. If index repair
@@ -313,6 +321,36 @@ class IndexedStorage<S extends Storable> implements Storage<S>, StorageAccess<S> return mQueryEngine.query(filter);
}
+ public void truncate() throws PersistException {
+ hasManagedIndexes: {
+ for (IndexInfo info : mAllIndexInfoMap.values()) {
+ if (info instanceof ManagedIndex) {
+ break hasManagedIndexes;
+ }
+ }
+
+ // No managed indexes, so nothing special to do.
+ mMasterStorage.truncate();
+ return;
+ }
+
+ Transaction txn = mRepository.enterTransaction();
+ try {
+ mMasterStorage.truncate();
+
+ // Now truncate the indexes.
+ for (IndexInfo info : mAllIndexInfoMap.values()) {
+ if (info instanceof ManagedIndex) {
+ ((ManagedIndex) info).getIndexEntryStorage().truncate();
+ }
+ }
+
+ txn.commit();
+ } finally {
+ txn.exit();
+ }
+ }
+
public boolean addTrigger(Trigger<? super S> trigger) {
return mMasterStorage.addTrigger(trigger);
}
@@ -372,9 +410,6 @@ class IndexedStorage<S extends Storable> implements Storage<S>, StorageAccess<S> }
}
- // FIXME: sort buffer should be on repository access. Also, create abstract
- // repository access that creates the correct merge sort buffer. And more:
- // create capability for managing merge sort buffers.
return new MergeSortBuffer<S>(mRootStorage);
}
@@ -521,20 +556,24 @@ class IndexedStorage<S extends Storable> implements Storage<S>, StorageAccess<S> {
// Doesn't completely remove the index, but it should free up space.
+
double desiredSpeed = mRepository.getIndexRepairThrottle();
Throttle throttle = desiredSpeed < 1.0 ? new Throttle(POPULATE_THROTTLE_WINDOW) : null;
- long totalDropped = 0;
- while (true) {
- Transaction txn = mRepository.getWrappedRepository()
- .enterTopTransaction(IsolationLevel.READ_COMMITTED);
- txn.setForUpdate(true);
- try {
- Cursor<? extends Storable> cursor = indexEntryStorage.query().fetch();
- if (!cursor.hasNext()) {
- break;
- }
- int count = 0;
+ if (throttle == null) {
+ indexEntryStorage.truncate();
+ } else {
+ long totalDropped = 0;
+ while (true) {
+ Transaction txn = mRepository.getWrappedRepository()
+ .enterTopTransaction(IsolationLevel.READ_COMMITTED);
+ txn.setForUpdate(true);
+ try {
+ Cursor<? extends Storable> cursor = indexEntryStorage.query().fetch();
+ if (!cursor.hasNext()) {
+ break;
+ }
+ int count = 0;
final long savedTotal = totalDropped;
boolean anyFailure = false;
try {
@@ -544,26 +583,25 @@ class IndexedStorage<S extends Storable> implements Storage<S>, StorageAccess<S> } else {
anyFailure = true;
}
+ }
+ } finally {
+ cursor.close();
+ }
+ txn.commit();
+ if (log.isInfoEnabled()) {
+ log.info("Removed " + totalDropped + " index entries");
}
- } finally {
- cursor.close();
- }
- txn.commit();
- if (log.isInfoEnabled()) {
- log.info("Removed " + totalDropped + " index entries");
- }
if (anyFailure && totalDropped <= savedTotal) {
log.warn("No indexes removed in last batch. " +
"Aborting index removal cleanup");
break;
}
- } catch (FetchException e) {
- throw e.toPersistException();
- } finally {
- txn.exit();
- }
+ } catch (FetchException e) {
+ throw e.toPersistException();
+ } finally {
+ txn.exit();
+ }
- if (throttle != null) {
try {
throttle.throttle(desiredSpeed, POPULATE_THROTTLE_SLEEP_PRECISION);
} catch (InterruptedException e) {
|