/*
|
* CDDL HEADER START
|
*
|
* The contents of this file are subject to the terms of the
|
* Common Development and Distribution License, Version 1.0 only
|
* (the "License"). You may not use this file except in compliance
|
* with the License.
|
*
|
* You can obtain a copy of the license at
|
* trunk/opends/resource/legal-notices/OpenDS.LICENSE
|
* or https://OpenDS.dev.java.net/OpenDS.LICENSE.
|
* See the License for the specific language governing permissions
|
* and limitations under the License.
|
*
|
* When distributing Covered Code, include this CDDL HEADER in each
|
* file and include the License file at
|
* trunk/opends/resource/legal-notices/OpenDS.LICENSE. If applicable,
|
* add the following below this CDDL HEADER, with the fields enclosed
|
* by brackets "[]" replaced with your own identifying information:
|
* Portions Copyright [yyyy] [name of copyright owner]
|
*
|
* CDDL HEADER END
|
*
|
*
|
* Portions Copyright 2006-2007 Sun Microsystems, Inc.
|
*/
|
package org.opends.server.backends.jeb;
|
import org.opends.messages.Message;
|
|
import org.opends.server.api.DirectoryThread;
|
import org.opends.server.loggers.debug.DebugTracer;
|
import static org.opends.server.loggers.debug.DebugLogger.getTracer;
|
import static org.opends.server.loggers.debug.DebugLogger.debugEnabled;
|
import static org.opends.server.loggers.ErrorLogger.logError;
|
import org.opends.server.admin.std.server.LocalDBBackendCfg;
|
import org.opends.server.types.*;
|
import org.opends.server.protocols.asn1.ASN1OctetString;
|
import static org.opends.server.util.StaticUtils.getFileForPath;
|
import static org.opends.messages.JebMessages.
|
INFO_JEB_INDEX_MERGE_NO_DATA;
|
import static org.opends.messages.JebMessages.
|
INFO_JEB_INDEX_MERGE_START;
|
import static org.opends.messages.JebMessages.
|
INFO_JEB_INDEX_MERGE_COMPLETE;
|
import java.util.*;
|
import java.io.*;
|
|
import com.sleepycat.je.Transaction;
|
|
/**
|
* A thread to merge a set of intermediate files from an vlvIndex builder
|
* into an vlvIndex database.
|
*/
|
public class VLVIndexMergeThread extends DirectoryThread
|
{
|
/**
|
* The tracer object for the debug logger.
|
*/
|
private static final DebugTracer TRACER = getTracer();
|
|
|
/**
|
* The buffer size to use when reading data from disk.
|
*/
|
private static final int INPUT_STREAM_BUFFER_SIZE = 65536;
|
|
/**
|
* The configuration of the JE backend containing the vlvIndex.
|
*/
|
LocalDBBackendCfg config;
|
|
/**
|
* The LDIF import configuration, which indicates whether we are
|
* appending to existing data.
|
*/
|
LDIFImportConfig ldifImportConfig;
|
|
/**
|
* The vlvIndex database being written.
|
*/
|
VLVIndex vlvIndex;
|
|
/**
|
* The name of the vlvIndex for use in file names and log messages.
|
*/
|
String indexName;
|
|
/**
|
* Indicates whether we are replacing existing data or not.
|
*/
|
private boolean replaceExisting = false;
|
|
private List<DataInputStream> addDataStreams;
|
private List<DataInputStream> delDataStreams;
|
|
/**
|
* A weak reference hash map used to cache last sort values read from files.
|
*/
|
private HashMap<DataInputStream,SortValues> lastAddValues =
|
new HashMap<DataInputStream,SortValues>();
|
|
private HashMap<DataInputStream,SortValues> lastDelValues =
|
new HashMap<DataInputStream,SortValues>();
|
|
|
/**
|
* A file name filter to identify temporary files we have written.
|
*/
|
private FilenameFilter filter = new FilenameFilter()
|
{
|
public boolean accept(File d, String name)
|
{
|
return name.startsWith(vlvIndex.getName());
|
}
|
};
|
|
/**
|
* Create a new vlvIndex merge thread.
|
* @param config The configuration of the JE backend containing the vlvIndex.
|
* @param ldifImportConfig The LDIF import configuration, which indicates
|
* whether we are appending to existing data.
|
* @param vlvIndex The vlvIndex database to be written.
|
*/
|
VLVIndexMergeThread(LocalDBBackendCfg config,
|
LDIFImportConfig ldifImportConfig,
|
VLVIndex vlvIndex)
|
{
|
super("Index Merge Thread " + vlvIndex.getName());
|
|
this.config = config;
|
this.ldifImportConfig = ldifImportConfig;
|
this.vlvIndex = vlvIndex;
|
replaceExisting =
|
ldifImportConfig.appendToExistingData() &&
|
ldifImportConfig.replaceExistingEntries();
|
addDataStreams = new ArrayList<DataInputStream>();
|
delDataStreams = new ArrayList<DataInputStream>();
|
lastAddValues = new HashMap<DataInputStream, SortValues>();
|
lastDelValues = new HashMap<DataInputStream, SortValues>();
|
}
|
|
/**
|
* Run this thread.
|
*/
|
public void run()
|
{
|
try
|
{
|
merge();
|
}
|
catch (Exception e)
|
{
|
if (debugEnabled())
|
{
|
TRACER.debugCaught(DebugLogLevel.ERROR, e);
|
}
|
|
throw new RuntimeException(e);
|
}
|
}
|
|
/**
|
* The merge phase builds the vlvIndex from intermediate files written
|
* during entry processing. Each line of an intermediate file has data for
|
* one vlvIndex key and the keys are in order. For each vlvIndex key, the data
|
* from each intermediate file containing a line for that key must be merged
|
* and written to the vlvIndex.
|
* @throws Exception If an error occurs.
|
*/
|
public void merge() throws Exception
|
{
|
// Open all the files.
|
File tempDir = getFileForPath(config.getImportTempDirectory());
|
File[] files = tempDir.listFiles(filter);
|
|
if (files == null || files.length == 0)
|
{
|
Message message = INFO_JEB_INDEX_MERGE_NO_DATA.get(vlvIndex.getName());
|
logError(message);
|
return;
|
}
|
|
if (debugEnabled())
|
{
|
Message message = INFO_JEB_INDEX_MERGE_START.get(
|
files.length, vlvIndex.getName());
|
TRACER.debugInfo(message.toString());
|
}
|
|
Transaction txn = null;
|
|
try
|
{
|
for (int i = 0; i < files.length; i++)
|
{
|
// Open a reader for this file.
|
BufferedInputStream bufferedStream =
|
new BufferedInputStream(new FileInputStream(files[i]),
|
INPUT_STREAM_BUFFER_SIZE);
|
DataInputStream dis = new DataInputStream(bufferedStream);
|
if(files[i].getName().endsWith("_add"))
|
{
|
addDataStreams.add(dis);
|
}
|
else if(files[i].getName().endsWith("_del"))
|
{
|
delDataStreams.add(dis);
|
}
|
}
|
|
while(true)
|
{
|
SortValuesSet currentSet = null;
|
SortValues maxKey = null;
|
// Get a set by using the smallest sort values
|
SortValues addValue = readNextAdd(maxKey);
|
|
// Process deletes first for this set
|
if(replaceExisting)
|
{
|
SortValues delValue = readNextDel(maxKey);
|
if(delValue != null)
|
{
|
if(currentSet == null)
|
{
|
if(addValue == null || delValue.compareTo(addValue) < 0)
|
{
|
// Set the current set using the del value.
|
currentSet = vlvIndex.getSortValuesSet(txn,
|
delValue.getEntryID(),
|
delValue.getValues());
|
}
|
else
|
{
|
// Set the current set using the add value.
|
currentSet = vlvIndex.getSortValuesSet(txn,
|
addValue.getEntryID(),
|
addValue.getValues());
|
}
|
maxKey = currentSet.getKeySortValues();
|
}
|
}
|
|
while(delValue != null)
|
{
|
currentSet.remove(delValue.getEntryID(), delValue.getValues());
|
delValue = readNextDel(maxKey);
|
}
|
}
|
|
if(addValue != null)
|
{
|
if(currentSet == null)
|
{
|
currentSet = vlvIndex.getSortValuesSet(txn, addValue.getEntryID(),
|
addValue.getValues());
|
maxKey = currentSet.getKeySortValues();
|
}
|
|
while(addValue != null)
|
{
|
currentSet.add(addValue.getEntryID(), addValue.getValues());
|
if(currentSet.size() > vlvIndex.getSortedSetCapacity())
|
{
|
// Need to split the set as it has exceeded the entry limit.
|
SortValuesSet splitSortValuesSet =
|
currentSet.split(currentSet.size() / 2);
|
// Find where the set split and see if the last added values
|
// is before or after the split.
|
SortValues newKey = currentSet.getKeySortValues();
|
|
if(debugEnabled())
|
{
|
TRACER.debugInfo("SortValuesSet with key %s has reached" +
|
" the entry size of %d. Spliting into two sets with " +
|
" keys %s and %s.", maxKey, currentSet.size(), newKey,
|
maxKey);
|
}
|
|
if(addValue.compareTo(newKey) < 0)
|
{
|
// The last added values is before the split so we have to
|
// keep adding to it.
|
vlvIndex.putSortValuesSet(txn, splitSortValuesSet);
|
maxKey = newKey;
|
}
|
else
|
{
|
// The last added values is after the split so we can add to
|
// the newly split set.
|
vlvIndex.putSortValuesSet(txn, currentSet);
|
currentSet = splitSortValuesSet;
|
}
|
}
|
addValue = readNextAdd(maxKey);
|
}
|
}
|
|
// We should have made all the modifications to this set. Store it back
|
// to database.
|
vlvIndex.putSortValuesSet(txn, currentSet);
|
|
if(maxKey == null)
|
{
|
// If we reached here, we should have processed all the sets and
|
// there should be nothing left to add or delete.
|
break;
|
}
|
}
|
|
if(!ldifImportConfig.appendToExistingData())
|
{
|
vlvIndex.setTrusted(txn, true);
|
}
|
}
|
finally
|
{
|
for(DataInputStream stream : addDataStreams)
|
{
|
stream.close();
|
}
|
|
for(DataInputStream stream : delDataStreams)
|
{
|
stream.close();
|
}
|
|
// Delete all the files.
|
if (files != null)
|
{
|
for (File f : files)
|
{
|
f.delete();
|
}
|
}
|
}
|
|
if (debugEnabled())
|
{
|
Message message = INFO_JEB_INDEX_MERGE_COMPLETE.get(vlvIndex.getName());
|
TRACER.debugInfo(message.toString());
|
}
|
}
|
|
/**
|
* Reads the next sort values from the files that is smaller then the max.
|
* @throws IOException If an I/O error occurs while reading the input file.
|
*/
|
private SortValues readNextAdd(SortValues maxValues)
|
throws IOException
|
{
|
for(DataInputStream dataInputStream : addDataStreams)
|
{
|
if(lastAddValues.get(dataInputStream) == null)
|
{
|
try
|
{
|
SortKey[] sortKeys = vlvIndex.sortOrder.getSortKeys();
|
EntryID id = new EntryID(dataInputStream.readLong());
|
AttributeValue[] attrValues =
|
new AttributeValue[sortKeys.length];
|
for(int i = 0; i < sortKeys.length; i++)
|
{
|
SortKey sortKey = sortKeys[i];
|
int length = dataInputStream.readInt();
|
if(length > 0)
|
{
|
byte[] valueBytes = new byte[length];
|
if(length == dataInputStream.read(valueBytes, 0, length))
|
{
|
attrValues[i] =
|
new AttributeValue(sortKey.getAttributeType(),
|
new ASN1OctetString(valueBytes));
|
}
|
}
|
|
}
|
lastAddValues.put(dataInputStream,
|
new SortValues(id, attrValues, vlvIndex.sortOrder));
|
}
|
catch (EOFException e)
|
{
|
continue;
|
}
|
}
|
}
|
|
Map.Entry<DataInputStream, SortValues> smallestEntry = null;
|
for(Map.Entry<DataInputStream, SortValues> entry :
|
lastAddValues.entrySet())
|
{
|
if(smallestEntry == null ||
|
entry.getValue().compareTo(smallestEntry.getValue()) < 0)
|
{
|
smallestEntry = entry;
|
}
|
}
|
|
if(smallestEntry != null)
|
{
|
SortValues smallestValues = smallestEntry.getValue();
|
if(maxValues == null || smallestValues.compareTo(maxValues) <= 0)
|
{
|
lastAddValues.remove(smallestEntry.getKey());
|
return smallestValues;
|
}
|
}
|
|
return null;
|
}
|
|
/**
|
* Reads the next sort values from the files that is smaller then the max.
|
* @throws IOException If an I/O error occurs while reading the input file.
|
*/
|
private SortValues readNextDel(SortValues maxValues)
|
throws IOException
|
{
|
for(DataInputStream dataInputStream : delDataStreams)
|
{
|
if(lastDelValues.get(dataInputStream) == null)
|
{
|
try
|
{
|
EntryID id = new EntryID(dataInputStream.readLong());
|
AttributeValue[] attrValues =
|
new AttributeValue[vlvIndex.sortOrder.getSortKeys().length];
|
int i = 0;
|
for(SortKey sortKey : vlvIndex.sortOrder.getSortKeys())
|
{
|
int length = dataInputStream.readInt();
|
if(length > 0)
|
{
|
byte[] valueBytes = new byte[length];
|
if(length == dataInputStream.read(valueBytes, 0, length))
|
{
|
attrValues[i] =
|
new AttributeValue(sortKey.getAttributeType(),
|
new ASN1OctetString(valueBytes));
|
}
|
}
|
}
|
lastDelValues.put(dataInputStream,
|
new SortValues(id, attrValues,
|
vlvIndex.sortOrder));
|
}
|
catch (EOFException e)
|
{
|
continue;
|
}
|
}
|
}
|
|
Map.Entry<DataInputStream, SortValues> smallestEntry = null;
|
for(Map.Entry<DataInputStream, SortValues> entry :
|
lastDelValues.entrySet())
|
{
|
if(smallestEntry == null ||
|
entry.getValue().compareTo(smallestEntry.getValue()) < 0)
|
{
|
smallestEntry = entry;
|
}
|
}
|
|
if(smallestEntry != null)
|
{
|
SortValues smallestValues = smallestEntry.getValue();
|
if(maxValues == null || smallestValues.compareTo(maxValues) <= 0)
|
{
|
lastDelValues.remove(smallestEntry.getKey());
|
return smallestValues;
|
}
|
}
|
|
return null;
|
}
|
}
|
