/*-
* See the file LICENSE for redistribution information.
*
* Copyright (c) 2002-2005
*      Sleepycat Software.  All rights reserved.
*
* $Id: DbScavenger.java,v 1.6 2004/12/22 14:11:38 linda Exp $
*/

package com.sleepycat.je.utilint;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.DbInternal;
import com.sleepycat.je.Environment;
import com.sleepycat.je.config.EnvironmentParams;
import com.sleepycat.je.dbi.DatabaseImpl;
import com.sleepycat.je.dbi.DbConfigManager;
import com.sleepycat.je.dbi.EnvironmentImpl;
import com.sleepycat.je.log.LastFileReader;
import com.sleepycat.je.log.LogEntryType;
import com.sleepycat.je.log.ScavengerFileReader;
import com.sleepycat.je.log.entry.LNLogEntry;
import com.sleepycat.je.log.entry.LogEntry;
import com.sleepycat.je.tree.LN;
import com.sleepycat.je.tree.MapLN;
import com.sleepycat.je.tree.NameLN;
import com.sleepycat.je.util.DbDump;

public class DbScavenger extends DbDump {

    private int readBufferSize;
    private EnvironmentImpl envImpl;

    /*
     * Set of committed txn ids that have been seen so far.
     */
    private Set committedTxnIdsSeen;

    /*
     * Set of LN Node Ids that have been seen so far.
     */
    private Set nodeIdsSeen;

    /*
     * Map of database id to database names.
     */
    private Map dbIdToName;

    /*
     * Map of database id to Boolean (dupSort).
     */
    private Map dbIdToDupSort;

    /*
     * Map of database id to the .dump file output stream for that database.
     */
    private Map dbIdToOutputStream;

    private boolean dumpCorruptedBounds = false;

    public DbScavenger(Environment env,
                       PrintStream outputFile,
                       boolean formatUsingPrintable,
                       boolean doAggressiveScavengerRun) {
	super(env, null, outputFile, formatUsingPrintable);

        this.doAggressiveScavengerRun = doAggressiveScavengerRun;
	this.dbIdToName = new HashMap();
	this.dbIdToDupSort = new HashMap();
	this.dbIdToOutputStream = new HashMap();
    }

    /**
     * Set to true if corrupted boundaries should be dumped out.
     */
    public void setDumpCorruptedBounds(boolean dumpCorruptedBounds) {
	this.dumpCorruptedBounds = dumpCorruptedBounds;
    }

    public void dump()
	throws IOException, DatabaseException {

	openEnv(false);

	envImpl = DbInternal.envGetEnvironmentImpl(env);
	DbConfigManager cm = envImpl.getConfigManager();
	try {
	    readBufferSize =
		cm.getInt(EnvironmentParams.LOG_ITERATOR_READ_SIZE);
	} catch (DatabaseException DBE) {
	    readBufferSize = 8192;
	}

	/*
	 * Find the end of the log.
	 */
	LastFileReader reader = new LastFileReader(envImpl, readBufferSize);
	while (reader.readNextEntry()) {
	}

	/* Tell the fileManager where the end of the log is. */
	long lastUsedLsn = reader.getLastValidLsn();
	long nextAvailableLsn = reader.getEndOfLog();
	envImpl.getFileManager().setLastPosition(nextAvailableLsn,
						 lastUsedLsn,
						 reader.getPrevOffset());

	/* Pass 1: Scavenge the dbtree. */
	scavengeDbTree(lastUsedLsn, nextAvailableLsn);
	/* Pass 2: Scavenge the databases. */
	scavenge(lastUsedLsn, nextAvailableLsn);
    }

    /*
     * Scan the log looking for records that are relevant for scavenging the db
     * tree.
     */
    private void scavengeDbTree(long lastUsedLsn, long nextAvailableLsn)
	throws IOException, DatabaseException {

	committedTxnIdsSeen = new HashSet();
	nodeIdsSeen = new HashSet();

	final ScavengerFileReader scavengerReader =
	    new ScavengerFileReader(envImpl, readBufferSize, lastUsedLsn,
				    DbLsn.NULL_LSN, nextAvailableLsn) {
		    protected void processEntryCallback(LogEntry entry,
							LogEntryType entryType)
			throws DatabaseException {

			processDbTreeEntry(entry, entryType);
		    }
		};

        scavengerReader.setTargetType(LogEntryType.LOG_MAPLN_TRANSACTIONAL);
        scavengerReader.setTargetType(LogEntryType.LOG_MAPLN);
        scavengerReader.setTargetType(LogEntryType.LOG_NAMELN_TRANSACTIONAL);
        scavengerReader.setTargetType(LogEntryType.LOG_NAMELN);
        scavengerReader.setTargetType(LogEntryType.LOG_TXN_COMMIT);
        scavengerReader.setTargetType(LogEntryType.LOG_TXN_ABORT);
	while (scavengerReader.readNextEntry()) {
	}
    }

    /*
     * Look at an entry and determine if it should be processed for scavenging.
     */
    private boolean checkProcessEntry(LogEntry entry,
				      LogEntryType entryType,
				      boolean pass2) {
	boolean isTransactional = entry.isTransactional();

	/* 
	 * If entry is txnal...
	 *  if a commit record, add to committed txn id set
	 *  if an abort record, ignore it and don't process.
	 *  if an LN, check if it's in the committed txn id set.
	 *     If it is, continue processing, otherwise ignore it.
	 */
	if (isTransactional) {
	    Long txnId = new Long(entry.getTransactionId());
	    if (entryType.equals(LogEntryType.LOG_TXN_COMMIT)) {
		committedTxnIdsSeen.add(txnId);
		/* No need to process this entry further. */
		return false;
	    }

	    if (entryType.equals(LogEntryType.LOG_TXN_ABORT)) {
		/* No need to process this entry further. */
		return false;
	    }

	    if (!committedTxnIdsSeen.contains(txnId)) {
		return false;
	    }
	}

	/*
	 * Check the nodeid to see if we've already seen it or not.
	 */
	if (entry instanceof LNLogEntry) {
	    LNLogEntry lnEntry = (LNLogEntry) entry;
	    LN ln = lnEntry.getLN();
	    Long nodeId = new Long(ln.getNodeId());
	    boolean isDelDupLN =
		entryType.equals(LogEntryType.
				 LOG_DEL_DUPLN_TRANSACTIONAL) ||
		entryType.equals(LogEntryType.LOG_DEL_DUPLN);

	    /* 
	     * If aggressive, don't worry about whether this node has been
	     * dumped already.
	     */
	    if (pass2 && doAggressiveScavengerRun) {
		return !isDelDupLN;
	    }
	    if (nodeIdsSeen.contains(nodeId)) {
		return false;
	    } else {
		nodeIdsSeen.add(nodeId);
		if (isDelDupLN) {

		    /*
		     * For deleted LN's, note the NodeId has having been
		     * processed, but, don't output them.
		     */
		    return false;
		} else {
		    return true;
		}
	    }
	}

	return false;
    }

    /*
     * Called once for each log entry during the pass 1 (scavenging the dbtree.
     */
    private void processDbTreeEntry(LogEntry entry, LogEntryType entryType)
	throws DatabaseException {

	boolean processThisEntry =
	    checkProcessEntry(entry, entryType, false);

	if (processThisEntry &&
	    (entry instanceof LNLogEntry)) {
	    LNLogEntry lnEntry = (LNLogEntry) entry;
	    LN ln = lnEntry.getLN();
	    if (ln instanceof NameLN) {
		String name = new String(lnEntry.getKey().getKey());
		Integer dbId = new Integer(((NameLN) ln).getId().getId());
		if (dbIdToName.containsKey(dbId) &&
		    !((String) dbIdToName.get(dbId)).equals(name)) {
		    throw new DatabaseException
			("Already name mapped for dbId: " + dbId +
			 " changed from " + (String) dbIdToName.get(dbId) +
			 " to " + name);
		} else {
		    dbIdToName.put(dbId, name);
		}
	    }

	    if (ln instanceof MapLN) {
		DatabaseImpl db = ((MapLN) ln).getDatabase();
		Integer dbId = new Integer(db.getId().getId());
		Boolean dupSort = Boolean.valueOf(db.getSortedDuplicates());
		if (dbIdToDupSort.containsKey(dbId)) {
		    throw new DatabaseException
			("Already saw dupSort entry for dbId: " + dbId);
		} else {
		    dbIdToDupSort.put(dbId, dupSort);
		}
	    }
	}
    }

    /*
     * Pass 2: scavenge the regular (non-dbtree) environment.
     */
    private void scavenge(long lastUsedLsn, long nextAvailableLsn)
	throws IOException, DatabaseException {

	final ScavengerFileReader scavengerReader =
	    new ScavengerFileReader(envImpl, readBufferSize, lastUsedLsn,
				    DbLsn.NULL_LSN, nextAvailableLsn) {
		    protected void processEntryCallback(LogEntry entry,
							LogEntryType entryType)
			throws DatabaseException {

			processRegularEntry(entry, entryType);
		    }
		};

        scavengerReader.setTargetType(LogEntryType.LOG_LN_TRANSACTIONAL);
        scavengerReader.setTargetType(LogEntryType.LOG_LN);
        scavengerReader.setTargetType
	    (LogEntryType.LOG_DEL_DUPLN_TRANSACTIONAL);
        scavengerReader.setTargetType(LogEntryType.LOG_DEL_DUPLN);
        scavengerReader.setTargetType(LogEntryType.LOG_TXN_COMMIT);
        scavengerReader.setTargetType(LogEntryType.LOG_TXN_ABORT);
	scavengerReader.setDumpCorruptedBounds(dumpCorruptedBounds);
	while (scavengerReader.readNextEntry()) {
	}
    }

    /*
     * Process an entry during pass 2.
     */
    private void processRegularEntry(LogEntry entry, LogEntryType entryType)
	throws DatabaseException {

	boolean processThisEntry =
	    checkProcessEntry(entry, entryType, true);

	if (processThisEntry) {
	    LNLogEntry lnEntry = (LNLogEntry) entry;
	    Integer dbId = new Integer(lnEntry.getDbId().getId());
	    PrintStream out = getOutputStream(dbId);
	    LN ln = lnEntry.getLN();
	    byte[] keyData = lnEntry.getKey().getKey();
	    byte[] data = ln.getData();
	    if (data != null &&
		data.length > 0) {
		dumpOne(out, keyData, formatUsingPrintable);
		dumpOne(out, data, formatUsingPrintable);
	    }
	}
    }

    /*
     * Return the output stream for the .dump file for database with id dbId.
     * If an output stream has not already been created, then create one.
     */
    private PrintStream getOutputStream(Integer dbId)
	throws DatabaseException {

	try {
	    PrintStream ret = (PrintStream) dbIdToOutputStream.get(dbId);
	    if (ret != null) {
		return ret;
	    }
	    String name = (String) dbIdToName.get(dbId);
	    if (name == null) {
		name = "db" + dbId;
	    }
	    File file = new File(envHome, name + ".dump");
           	    ret = new PrintStream(new FileOutputStream(file));
	    dbIdToOutputStream.put(dbId, ret);
	    Boolean dupSort = (Boolean) dbIdToDupSort.get(dbId);
	    if (dupSort == null) {
		dupSort = Boolean.valueOf(false);
	    }
	    printHeader(ret, dupSort.booleanValue(), formatUsingPrintable);
	    return ret;
	} catch (IOException IOE) {
	    throw new DatabaseException(IOE);
	}
    }
}
