/*
|
* 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.server.protocols.asn1.ASN1OctetString;
|
import org.opends.server.types.Entry;
|
|
import com.sleepycat.je.DatabaseException;
|
import com.sleepycat.je.Transaction;
|
|
import java.util.Arrays;
|
import java.util.ArrayList;
|
import java.util.Collections;
|
import java.util.HashSet;
|
import java.util.Set;
|
import java.io.ByteArrayOutputStream;
|
import java.io.BufferedOutputStream;
|
import java.io.DataOutputStream;
|
import java.io.File;
|
import java.io.FilenameFilter;
|
import java.io.FileOutputStream;
|
import java.io.IOException;
|
|
/**
|
* This class is used to create an attribute index for an import process.
|
* It is used as follows.
|
* <pre>
|
* startProcessing();
|
* processEntry(entry);
|
* processEntry(entry);
|
* ...
|
* stopProcessing();
|
* merge();
|
* </pre>
|
*/
|
public class IndexBuilder
|
{
|
/**
|
* The import context.
|
*/
|
private ImportContext importContext;
|
|
/**
|
* The index database.
|
*/
|
private Index index;
|
|
/**
|
* The indexer to generate the index keys.
|
*/
|
private Indexer indexer;
|
|
/**
|
* The write buffer.
|
*/
|
ArrayList<IndexMod> buffer;
|
|
/**
|
* The write buffer size.
|
*/
|
private int bufferSize;
|
|
/**
|
* Current output file number.
|
*/
|
private int fileNumber = 0;
|
|
/**
|
* The index entry limit.
|
*/
|
private int entryLimit;
|
|
/**
|
* A unique prefix for temporary files to prevent conflicts.
|
*/
|
private String fileNamePrefix;
|
|
/**
|
* Indicates whether we are replacing existing data or not.
|
*/
|
private boolean replaceExisting = false;
|
|
|
private ByteArrayOutputStream addBytesStream = new ByteArrayOutputStream();
|
private ByteArrayOutputStream delBytesStream = new ByteArrayOutputStream();
|
|
private DataOutputStream addBytesDataStream;
|
private DataOutputStream delBytesDataStream;
|
|
/**
|
* 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(fileNamePrefix);
|
}
|
};
|
|
/**
|
* Construct an index builder.
|
*
|
* @param importContext The import context.
|
* @param index The index database we are writing.
|
* @param entryLimit The index entry limit.
|
* @param bufferSize The amount of memory available for buffering.
|
*/
|
public IndexBuilder(ImportContext importContext,
|
Index index, int entryLimit, long bufferSize)
|
{
|
this.importContext = importContext;
|
this.index = index;
|
this.indexer = index.indexer;
|
this.entryLimit = entryLimit;
|
this.bufferSize = (int)bufferSize/100;
|
long tid = Thread.currentThread().getId();
|
fileNamePrefix = importContext.getContainerName() + "_" +
|
indexer.toString() + "_" + tid + "_";
|
replaceExisting =
|
importContext.getLDIFImportConfig().appendToExistingData() &&
|
importContext.getLDIFImportConfig().replaceExistingEntries();
|
addBytesDataStream = new DataOutputStream(addBytesStream);
|
delBytesDataStream = new DataOutputStream(delBytesStream);
|
}
|
|
/**
|
* This method must be called before this object can process any
|
* entries. It cleans up any temporary files left over from a
|
* previous import.
|
*/
|
public void startProcessing()
|
{
|
// Clean up any work files left over from a previous run.
|
File tempDir = new File(importContext.getConfig().getImportTempDirectory());
|
File[] files = tempDir.listFiles(filter);
|
if (files != null)
|
{
|
for (File f : files)
|
{
|
f.delete();
|
}
|
}
|
|
buffer = new ArrayList<IndexMod>(bufferSize);
|
}
|
|
/**
|
* Indicates that the index thread should process the provided entry.
|
* @param oldEntry The existing contents of the entry, or null if this is
|
* a new entry.
|
* @param newEntry The new contents of the entry.
|
* @param entryID The entry ID.
|
* @throws DatabaseException If an error occurs in the JE database.
|
* @throws IOException If an I/O error occurs while writing an intermediate
|
* file.
|
*/
|
public void processEntry(Entry oldEntry, Entry newEntry, EntryID entryID)
|
throws DatabaseException, IOException
|
{
|
Transaction txn = null;
|
|
// Update the index for this entry.
|
if (oldEntry != null)
|
{
|
// This is an entry being replaced.
|
Set<ASN1OctetString> addKeys = new HashSet<ASN1OctetString>();
|
Set<ASN1OctetString> delKeys = new HashSet<ASN1OctetString>();
|
|
indexer.replaceEntry(txn, oldEntry, newEntry, addKeys, delKeys);
|
|
for (ASN1OctetString k : delKeys)
|
{
|
removeID(k.value(), entryID);
|
}
|
|
for (ASN1OctetString k : addKeys)
|
{
|
insertID(k.value(), entryID);
|
}
|
}
|
else
|
{
|
// This is a new entry.
|
Set<ASN1OctetString> addKeys = new HashSet<ASN1OctetString>();
|
indexer.indexEntry(txn, newEntry, addKeys);
|
for (ASN1OctetString k : addKeys)
|
{
|
insertID(k.value(), entryID);
|
}
|
}
|
|
}
|
|
|
|
/**
|
* Indicates that there will be no more updates.
|
* @throws IOException If an I/O error occurs while writing an intermediate
|
* file.
|
*/
|
public void stopProcessing() throws IOException
|
{
|
flushBuffer();
|
}
|
|
|
|
/**
|
* Get a statistic of the number of keys that reached the entry limit.
|
*
|
* @return The number of keys that reached the entry limit.
|
*/
|
public int getEntryLimitExceededCount()
|
{
|
return index.getEntryLimitExceededCount();
|
}
|
|
/**
|
* Record the insertion of an entry ID.
|
* @param key The index key.
|
* @param entryID The entry ID.
|
* @throws IOException If an I/O error occurs while writing an intermediate
|
* file.
|
*/
|
private void insertID(byte[] key, EntryID entryID)
|
throws IOException
|
{
|
if (buffer.size() >= bufferSize)
|
{
|
flushBuffer();
|
}
|
|
IndexMod kav = new IndexMod(key, entryID, false);
|
buffer.add(kav);
|
}
|
|
/**
|
* Record the deletion of an entry ID.
|
* @param key The index key.
|
* @param entryID The entry ID.
|
* @throws IOException If an I/O error occurs while writing an intermediate
|
* file.
|
*/
|
private void removeID(byte[] key, EntryID entryID)
|
throws IOException
|
{
|
if (buffer.size() >= bufferSize)
|
{
|
flushBuffer();
|
}
|
|
IndexMod kav = new IndexMod(key, entryID, true);
|
buffer.add(kav);
|
}
|
|
/**
|
* Called when the buffer is full. It first sorts the buffer using the same
|
* key comparator used by the index database. Then it merges all the
|
* IDs for the same key together and writes each key and its list of IDs
|
* to an intermediate binary file.
|
* A list of deleted IDs is only present if we are replacing existing entries.
|
*
|
* @throws IOException If an I/O error occurs while writing an intermediate
|
* file.
|
*/
|
private void flushBuffer() throws IOException
|
{
|
if (buffer.size() == 0)
|
{
|
return;
|
}
|
|
// Keys must be sorted before we can merge duplicates.
|
IndexModComparator comparator;
|
if (replaceExisting)
|
{
|
// The entry IDs may be out of order.
|
// We must sort by key and ID.
|
comparator = new IndexModComparator(indexer.getComparator(), true);
|
}
|
else
|
{
|
// The entry IDs are all new and are therefore already ordered.
|
// We just need to sort by key.
|
comparator = new IndexModComparator(indexer.getComparator(), false);
|
}
|
Collections.sort(buffer, comparator);
|
|
// Start a new file.
|
fileNumber++;
|
String fileName = fileNamePrefix + String.valueOf(fileNumber);
|
File file = new File(importContext.getConfig().getImportTempDirectory(),
|
fileName);
|
BufferedOutputStream bufferedStream =
|
new BufferedOutputStream(new FileOutputStream(file));
|
DataOutputStream dataStream = new DataOutputStream(bufferedStream);
|
|
// Reset the byte array output streams but preserve the underlying arrays.
|
addBytesStream.reset();
|
delBytesStream.reset();
|
|
try
|
{
|
byte[] currentKey = null;
|
for (IndexMod key : buffer)
|
{
|
byte[] keyString = key.key;
|
if (!Arrays.equals(keyString,currentKey))
|
{
|
if (currentKey != null)
|
{
|
dataStream.writeInt(currentKey.length);
|
dataStream.write(currentKey);
|
dataStream.writeInt(addBytesStream.size());
|
addBytesStream.writeTo(dataStream);
|
if (replaceExisting)
|
{
|
dataStream.writeInt(delBytesStream.size());
|
delBytesStream.writeTo(dataStream);
|
}
|
}
|
|
currentKey = keyString;
|
addBytesStream.reset();
|
delBytesStream.reset();
|
}
|
|
if (key.isDelete)
|
{
|
delBytesDataStream.writeLong(key.value.longValue());
|
}
|
else
|
{
|
addBytesDataStream.writeLong(key.value.longValue());
|
}
|
|
}
|
|
if (currentKey != null)
|
{
|
dataStream.writeInt(currentKey.length);
|
dataStream.write(currentKey);
|
dataStream.writeInt(addBytesStream.size());
|
addBytesStream.writeTo(dataStream);
|
if (replaceExisting)
|
{
|
dataStream.writeInt(delBytesStream.size());
|
delBytesStream.writeTo(dataStream);
|
}
|
}
|
|
buffer = new ArrayList<IndexMod>(bufferSize);
|
}
|
finally
|
{
|
dataStream.close();
|
}
|
}
|
|
/**
|
* Get a string that identifies this index builder.
|
*
|
* @return A string that identifies this index builder.
|
*/
|
public String toString()
|
{
|
return indexer.toString();
|
}
|
}
|
