Skip to content

Commit

Permalink
GEODE-2643: Combine chunk and file region into a single region
Browse files Browse the repository at this point in the history 
* removed file and chunk count from stat
* removed tests that were doing checks against chunk bucketRegions
  • Loading branch information
@jhuynh1
jhuynh1 committed Mar 14, 2017
1 parent 3c212fb commit d8a8973
Show file tree
Hide file tree
Showing 17 changed files with 131 additions and 382 deletions.
28 changes: 13 additions & 15 deletions ...e-lucene/src/main/java/org/apache/geode/cache/lucene/internal/IndexRepositoryFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,20 +43,18 @@ public IndexRepository computeIndexRepository(final Integer bucketId, LuceneSeri
LuceneIndexImpl index, PartitionedRegion userRegion, final IndexRepository oldRepository)
throws IOException {
LuceneIndexForPartitionedRegion indexForPR = (LuceneIndexForPartitionedRegion) index;
final PartitionedRegion fileRegion = indexForPR.getFileRegion();
final PartitionedRegion chunkRegion = indexForPR.getChunkRegion();
final PartitionedRegion fileRegion = indexForPR.getFileAndChunkRegion();

BucketRegion fileBucket = getMatchingBucket(fileRegion, bucketId);
BucketRegion chunkBucket = getMatchingBucket(chunkRegion, bucketId);
BucketRegion fileAndChunkBucket = getMatchingBucket(fileRegion, bucketId);
BucketRegion dataBucket = getMatchingBucket(userRegion, bucketId);
boolean success = false;
if (fileBucket == null || chunkBucket == null) {
if (fileAndChunkBucket == null) {
if (oldRepository != null) {
oldRepository.cleanup();
}
return null;
}
if (!chunkBucket.getBucketAdvisor().isPrimary()) {
if (!fileAndChunkBucket.getBucketAdvisor().isPrimary()) {
if (oldRepository != null) {
oldRepository.cleanup();
}
Expand All @@ -71,26 +69,26 @@ public IndexRepository computeIndexRepository(final Integer bucketId, LuceneSeri
oldRepository.cleanup();
}
DistributedLockService lockService = getLockService();
String lockName = getLockName(bucketId, fileBucket);
String lockName = getLockName(fileAndChunkBucket);
while (!lockService.lock(lockName, 100, -1)) {
if (!chunkBucket.getBucketAdvisor().isPrimary()) {
if (!fileAndChunkBucket.getBucketAdvisor().isPrimary()) {
return null;
}
}

final IndexRepository repo;
try {
RegionDirectory dir = new RegionDirectory(getBucketTargetingMap(fileBucket, bucketId),
getBucketTargetingMap(chunkBucket, bucketId), indexForPR.getFileSystemStats());
RegionDirectory dir = new RegionDirectory(getBucketTargetingMap(fileAndChunkBucket, bucketId),
indexForPR.getFileSystemStats());
IndexWriterConfig config = new IndexWriterConfig(indexForPR.getAnalyzer());
IndexWriter writer = new IndexWriter(dir, config);
repo = new IndexRepositoryImpl(fileBucket, writer, serializer, indexForPR.getIndexStats(),
dataBucket, lockService, lockName);
repo = new IndexRepositoryImpl(fileAndChunkBucket, writer, serializer,
indexForPR.getIndexStats(), dataBucket, lockService, lockName);
success = true;
return repo;
} catch (IOException e) {
logger.info("Exception thrown while constructing Lucene Index for bucket:" + bucketId
+ " for file region:" + fileBucket.getFullPath());
+ " for file region:" + fileAndChunkBucket.getFullPath());
throw e;
} finally {
if (!success) {
Expand All @@ -104,8 +102,8 @@ private Map getBucketTargetingMap(BucketRegion region, int bucketId) {
return new BucketTargetingMap(region, bucketId);
}

private String getLockName(final Integer bucketId, final BucketRegion fileBucket) {
return FILE_REGION_LOCK_FOR_BUCKET_ID + fileBucket.getFullPath();
private String getLockName(final BucketRegion fileAndChunkBucket) {
return FILE_REGION_LOCK_FOR_BUCKET_ID + fileAndChunkBucket.getFullPath();
}

private DistributedLockService getLockService() {
Expand Down
76 changes: 20 additions & 56 deletions ...src/main/java/org/apache/geode/cache/lucene/internal/LuceneIndexForPartitionedRegion.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.lucene.internal.directory.DumpDirectoryFiles;
import org.apache.geode.cache.lucene.internal.filesystem.ChunkKey;
import org.apache.geode.cache.lucene.internal.filesystem.File;
import org.apache.geode.cache.lucene.internal.filesystem.FileSystemStats;
import org.apache.geode.cache.lucene.internal.partition.BucketTargetingFixedResolver;
Expand All @@ -48,12 +47,10 @@

/* wrapper of IndexWriter */
public class LuceneIndexForPartitionedRegion extends LuceneIndexImpl {
protected Region<String, File> fileRegion;
protected Region<ChunkKey, byte[]> chunkRegion;
protected Region fileAndChunkRegion;
protected final FileSystemStats fileSystemStats;

public static final String FILES_REGION_SUFFIX = ".files";
public static final String CHUNKS_REGION_SUFFIX = ".chunks";

public LuceneIndexForPartitionedRegion(String indexName, String regionPath, Cache cache) {
super(indexName, regionPath, cache);
Expand Down Expand Up @@ -81,16 +78,12 @@ protected RepositoryManager createRepositoryManager() {
regionShortCut = RegionShortcut.PARTITION;
}

// create PR fileRegion, but not to create its buckets for now
// create PR fileAndChunkRegion, but not to create its buckets for now
final String fileRegionName = createFileRegionName();
PartitionAttributes partitionAttributes = dataRegion.getPartitionAttributes();
if (!fileRegionExists(fileRegionName)) {
fileRegion =
createFileRegion(regionShortCut, fileRegionName, partitionAttributes, regionAttributes);
}


// create PR chunkRegion, but not to create its buckets for now
final String chunkRegionName = createChunkRegionName();

// we will create RegionDirectories on the fly when data comes in
HeterogeneousLuceneSerializer mapper = new HeterogeneousLuceneSerializer(getFieldNames());
Expand All @@ -99,59 +92,41 @@ protected RepositoryManager createRepositoryManager() {
DM dm = ((GemFireCacheImpl) getCache()).getDistributedSystem().getDistributionManager();
LuceneBucketListener lucenePrimaryBucketListener =
new LuceneBucketListener(partitionedRepositoryManager, dm);
if (!chunkRegionExists(chunkRegionName)) {
chunkRegion = createChunkRegion(regionShortCut, fileRegionName, partitionAttributes,
chunkRegionName, regionAttributes, lucenePrimaryBucketListener);

if (!fileRegionExists(fileRegionName)) {
fileAndChunkRegion = createFileRegion(regionShortCut, fileRegionName, partitionAttributes,
regionAttributes, lucenePrimaryBucketListener);
}
fileSystemStats.setFileSupplier(() -> (int) getFileRegion().getLocalSize());
fileSystemStats.setChunkSupplier(() -> (int) getChunkRegion().getLocalSize());
fileSystemStats.setBytesSupplier(() -> getChunkRegion().getPrStats().getDataStoreBytesInUse());

return partitionedRepositoryManager;
}
fileSystemStats
.setBytesSupplier(() -> getFileAndChunkRegion().getPrStats().getDataStoreBytesInUse());

public PartitionedRegion getFileRegion() {
return (PartitionedRegion) fileRegion;
return partitionedRepositoryManager;
}

public PartitionedRegion getChunkRegion() {
return (PartitionedRegion) chunkRegion;
public PartitionedRegion getFileAndChunkRegion() {
return (PartitionedRegion) fileAndChunkRegion;
}

public FileSystemStats getFileSystemStats() {
return fileSystemStats;
}

boolean fileRegionExists(String fileRegionName) {
return cache.<String, File>getRegion(fileRegionName) != null;
return cache.getRegion(fileRegionName) != null;
}

Region createFileRegion(final RegionShortcut regionShortCut, final String fileRegionName,
final PartitionAttributes partitionAttributes, final RegionAttributes regionAttributes) {
final PartitionAttributes partitionAttributes, final RegionAttributes regionAttributes,
PartitionListener listener) {
return createRegion(fileRegionName, regionShortCut, this.regionPath, partitionAttributes,
regionAttributes, null);
regionAttributes, listener);
}

public String createFileRegionName() {
return LuceneServiceImpl.getUniqueIndexRegionName(indexName, regionPath, FILES_REGION_SUFFIX);
}

boolean chunkRegionExists(String chunkRegionName) {
return cache.<ChunkKey, byte[]>getRegion(chunkRegionName) != null;
}

Region<ChunkKey, byte[]> createChunkRegion(final RegionShortcut regionShortCut,
final String fileRegionName, final PartitionAttributes partitionAttributes,
final String chunkRegionName, final RegionAttributes regionAttributes,
final PartitionListener lucenePrimaryBucketListener) {
return createRegion(chunkRegionName, regionShortCut, fileRegionName, partitionAttributes,
regionAttributes, lucenePrimaryBucketListener);
}

public String createChunkRegionName() {
return LuceneServiceImpl.getUniqueIndexRegionName(indexName, regionPath, CHUNKS_REGION_SUFFIX);
}

private PartitionAttributesFactory configureLuceneRegionAttributesFactory(
PartitionAttributesFactory attributesFactory,
PartitionAttributes<?, ?> dataRegionAttributes) {
Expand All @@ -175,8 +150,7 @@ protected <K, V> Region<K, V> createRegion(final String regionName,
final RegionShortcut regionShortCut, final String colocatedWithRegionName,
final PartitionAttributes partitionAttributes, final RegionAttributes regionAttributes,
PartitionListener lucenePrimaryBucketListener) {
PartitionAttributesFactory partitionAttributesFactory =
new PartitionAttributesFactory<String, File>();
PartitionAttributesFactory partitionAttributesFactory = new PartitionAttributesFactory();
if (lucenePrimaryBucketListener != null) {
partitionAttributesFactory.addPartitionListener(lucenePrimaryBucketListener);
}
Expand Down Expand Up @@ -217,23 +191,13 @@ public void destroy(boolean initiator) {
// Invoke super destroy to remove the extension
super.destroy(initiator);

// Destroy the chunk region (colocated with the file region)
// localDestroyRegion can't be used because locally destroying regions is not supported on
// colocated regions
if (!chunkRegion.isDestroyed()) {
chunkRegion.destroyRegion();
if (logger.isDebugEnabled()) {
logger.debug("Destroyed chunkRegion=" + chunkRegion.getName());
}
}

// Destroy the file region (colocated with the application region)
// localDestroyRegion can't be used because locally destroying regions is not supported on
// colocated regions
if (!fileRegion.isDestroyed()) {
fileRegion.destroyRegion();
if (!fileAndChunkRegion.isDestroyed()) {
fileAndChunkRegion.destroyRegion();
if (logger.isDebugEnabled()) {
logger.debug("Destroyed fileRegion=" + fileRegion.getName());
logger.debug("Destroyed fileAndChunkRegion=" + fileAndChunkRegion.getName());
}
}

Expand Down
5 changes: 2 additions & 3 deletions ...ucene/src/main/java/org/apache/geode/cache/lucene/internal/directory/RegionDirectory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,10 +47,9 @@ public class RegionDirectory extends BaseDirectory {
* Create a region directory with a given file and chunk region. These regions may be bucket
* regions or they may be replicated regions.
*/
public RegionDirectory(Map<String, File> fileRegion, Map<ChunkKey, byte[]> chunkRegion,
FileSystemStats stats) {
public RegionDirectory(Map fileAndChunkRegion, FileSystemStats stats) {
super(new SingleInstanceLockFactory());
fs = new FileSystem(fileRegion, chunkRegion, stats);
fs = new FileSystem(fileAndChunkRegion, stats);
}

@Override
Expand Down
Loading

0 comments on commit d8a8973

Please sign in to comment.