versioningrdbmssail.java

这是外国一个开源推理机

/*  OMM - Ontology Middleware Module
*  Copyright (C) 2002 OntoText Lab, Sirma AI OOD
*
*  Contact:
*  Sirma AI OOD, OntoText Lab.
*  38A, Christo Botev Blvd.
*  1000 Sofia, Bulgaria
*  tel. [phone]
*  fax. [phone]
*  [e-mail]
*
*  http://www.ontotext.com/
*
*  This library is free software; you can redistribute it and/or
*  modify it under the terms of the GNU Lesser General Public
*  License as published by the Free Software Foundation; either
*  version 2.1 of the License, or (at your option) any later version.
*
*  This library is distributed in the hope that it will be useful,
*  but WITHOUT ANY WARRANTY; without even the implied warranty of
*  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
*  Lesser General Public License for more details.
*
*  You should have received a copy of the GNU Lesser General Public
*  License along with this library; if not, write to the Free Software
*  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
package org.openrdf.sesame.sailimpl.omm.versioning;
/**
 @todo: short notes about the main idea
 * The general idea is to handle only start/commitTransaction and to shiff about
 * changes in Triples table related to explicit statements only - if such occur
 * we should update all _HIST tables among the creation of a new stateID to
 * associate all changes with it.
 *
 **/
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.openrdf.util.log.ThreadLog;

import org.openrdf.model.Literal;
import org.openrdf.model.Resource;
import org.openrdf.model.URI;
import org.openrdf.model.Value;

import org.openrdf.sesame.config.AccessDeniedException;
import org.openrdf.sesame.config.ConfigurationException;
import org.openrdf.sesame.config.RepositoryConfig;
import org.openrdf.sesame.config.SailConfig;
import org.openrdf.sesame.config.SystemConfig;
import org.openrdf.sesame.config.UnknownRepositoryException;
import org.openrdf.sesame.config.UserInfo;
import org.openrdf.sesame.omm.SessionContext;
import org.openrdf.sesame.omm.VersionManagement;
import org.openrdf.sesame.repository.local.LocalRepository;
import org.openrdf.sesame.repository.local.LocalService;
import org.openrdf.sesame.sail.NamespaceIterator;
import org.openrdf.sesame.sail.RdfRepository;
import org.openrdf.sesame.sail.Sail;
import org.openrdf.sesame.sail.SailInitializationException;
import org.openrdf.sesame.sail.SailInternalException;
import org.openrdf.sesame.sail.SailUpdateException;
import org.openrdf.sesame.sailimpl.rdbms.RDBMS;
import org.openrdf.sesame.sailimpl.rdbms.RdfSchemaRepository;
import org.openrdf.sesame.server.SesameServer;

/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2003</p>
 * <p>Company: </p>
 * @author unascribed
 * @version 1.0
 */
public class VersioningRdbmsSail
    extends RdfSchemaRepository implements VersionManagement {

  public VersioningRdbmsSail() {
  }

  protected ArrayList createdRepositories = new ArrayList();
  protected Connection adminConnection = null;

  /**
   * this methos DROPs all temporary brenches of the repository. invoked either
   * in clearRepository() or whitin shutDown()
   */
  protected void dropBranchedVersions()
  {
    if (adminConnection != null)
    {
      ThreadLog.trace("adminConnection != null");

      Iterator iter = createdRepositories.iterator();
      try {
        java.sql.Statement st = adminConnection.createStatement();
        while (iter.hasNext())
        {
          String branchedDatabaseName = (String)iter.next();
          ThreadLog.trace("branchedDatabaseName ");
          try {
            st.executeQuery("DROP DATABASE "+branchedDatabaseName);
            ThreadLog.trace("ok\n");
          } catch (SQLException ex) {
            ThreadLog.trace("false "+ex.getMessage()+"\n");
          }
        }
      } catch(SQLException e) {
        ThreadLog.trace("SQLException "+e.getMessage()+"\n");
      }
    }
    createdRepositories.clear();
  }

  /**
   * Allow the SAIL to synchronize any stale data. The SAIL can assume that
   * shutDown() is called before an application is stopped.
   *
   * @exception SailInternalException To indicate an internal error.
   */
  public void shutDown()
  {
    ThreadLog.trace("VersioningRdbmsSail ->shutDown(begin)");
    dropBranchedVersions();
    super.shutDown();
    ThreadLog.trace("VersioningRdbmsSail ->shutDown(end)");
  }

  protected int VersionUid;
  protected int baseUrlIndex;

  /**
   * Initializes the TmsSQL92Sail. The supplied Map should contain the
   * following key-value pairs:
   * <table>
   * <tr><td>key</td><td>value</td></tr>
   * <tr><td>jdbcDriver</td><td>The String representing the JDBC-driver
   * class, e.g. "org.gjt.mm.mysql.Driver"</td></tr>
   * <tr><td>jdbcUrl</td><td>The String representing the JDBC-url of the
   * database to connect to, e.g. "jdbc:mysql://localhost/sesame"</td></tr>
   * <tr><td>user</td><td>The username that can be used to connect to the
   * DBMS</td></tr>
   * <tr><td>password</td><td>The password of the user in the DBMS</td></tr>
   * </table>
   *
   * @param configParams configuration parameters
   * @exception SailInitializationException If the TmsSQL92Sail could not
   * be initialized using the supplied parameters.
   */
  public void initialize(Map configParams)
      throws SailInitializationException
  {
    ThreadLog.trace("initializing VersioningRdbmsSail");

    super.initialize(configParams);

    ThreadLog.trace("VersioningRdbmsSail initialized");
  }

 /*----------------------------------------+
   | Database initialization                 |
   +----------------------------------------*/

   /**
    * Initializes the database. _initDatabase() creates tables, indexes and
    * inserts default values into the database.
    **/
  protected void _initDatabase()
      throws SailInitializationException
  {
    try {
      super._initDatabase();
      // EXPECTED_STATEMENTS_TABLE table
      createExpectedSIDSTable();

      // BaseUrlTable table
      createBaseUrlTable();

      // Updates table
      createUpdatesTable();

      // TripleHIST table
      createTripleHIST_Table();

      // versions table
      createVersionTable();

      // active states table
      createActiveStatesTable();
    }
    catch (SQLException e) {
      throw new SailInitializationException(e);
    }
  } //initDatabase

  private static final String SUFFIX = "_HIST";
  protected static final String UPDATES_TABLE = "updates";
  protected static final String VERSION_TABLE = "version";
  protected static final String TRIPLES_HIST_TABLE = TRIPLES_TABLE+SUFFIX;
  protected static final String BASEURL_TABLE = "baseurls";
  protected static final String EXPECTED_STATEMENTS_TABLE = "expected_sids";

  protected static final String ACTIVE_STATES_TABLE = "active";

  //  SQLTables
  /**
   * we need a unique suffix for the branched states of the repository so this help us to
   * reuse the same sufix everywhere
   * @param state is the stateID which suffix to get
   * @return a string suffix
   */
  private String getSuffix(int state)
  {
    String result = "";
    result+="_"+state;
    return result;
  }

  protected RDBMS RDBMS() {
      return _rdbms;
  }

  protected boolean createExpectedSIDSTable()
      throws SQLException
  {
    if (!RDBMS().tableExists(EXPECTED_STATEMENTS_TABLE)) {
      String tableQ = "CREATE TABLE "+EXPECTED_STATEMENTS_TABLE+" ("+
                      "sid " + RDBMS().ID_INT + " NOT NULL,"+
                      "found " + RDBMS().BOOLEAN + ")";
      RDBMS().executeUpdate(tableQ);
      return true;
    }
    return false;
  }


  protected boolean createBaseUrlTable()
      throws SQLException
  {
    if (!RDBMS().tableExists(BASEURL_TABLE)) {
      String tableQ = "CREATE TABLE "+BASEURL_TABLE+" ("+
                      "urlID " + RDBMS().ID_INT + " NOT NULL,"+
                      "url " + RDBMS().LOCALNAME + ")";
      RDBMS().executeUpdate(tableQ);
      return true;
    }
    return false;
  }

  protected boolean createUpdatesTable()
      throws SQLException
  {
    if (!RDBMS().tableExists(UPDATES_TABLE)) {
      String tableQ = "CREATE TABLE "+UPDATES_TABLE+" ("+
                      "uid " + RDBMS().ID_INT + " NOT NULL,"+
                      "time TIMESTAMP ,"+
                      "usid " + RDBMS().ID_INT + " ,"+
                      "kind " + RDBMS().BOOLEAN + " , "+
                      "urlID " + RDBMS().ID_INT + ")";
      RDBMS().executeUpdate(tableQ);

      RDBMS().createIndex(UPDATES_TABLE, new String[] {"uid"}, true);
      return true;
    }
    return false;
  }

  // TripleHIST table
  protected boolean createTripleHIST_Table()
      throws SQLException
  {
    if (!RDBMS().tableExists(TRIPLES_HIST_TABLE)) {
      String tableQ = "CREATE TABLE "+TRIPLES_HIST_TABLE+" ("+
                      "id " + RDBMS().ID_INT + " NOT NULL,"+
                      "subject " + RDBMS().ID_INT + " NOT NULL REFERENCES " +
                      RESOURCES_TABLE + "(id), " +
                      "predicate " + RDBMS().ID_INT + " NOT NULL REFERENCES " +
                      RESOURCES_TABLE + "(id), " +
                      "object " + RDBMS().ID_INT + " NOT NULL, " +
                      "explicit " + RDBMS().BOOLEAN + " NOT NULL, " +
                      "BornAt " + RDBMS().ID_INT + " , "+
                      "DiedAt " + RDBMS().ID_INT + " )";
      RDBMS().executeUpdate(tableQ);

      RDBMS().createIndex(TRIPLES_HIST_TABLE, new String[]{"id"}, false);
      return true;
    }
    return false;
  } // createTripleHIST_Table()

  // Version Table
  protected boolean createVersionTable()
      throws SQLException
  {
    if (!RDBMS().tableExists(VERSION_TABLE)) {
      String tableQ = "CREATE TABLE "+VERSION_TABLE+" ("+
                      "vid " + RDBMS().ID_INT + " NOT NULL,"+
                      "usid " + RDBMS().ID_INT + " , "+
                      "uid " + RDBMS().ID_INT + " , "+
                      "label VARCHAR(255) )";
      RDBMS().executeUpdate(tableQ);

      RDBMS().createIndex(VERSION_TABLE, new String[]{"vid"}, true);
      return true;
    }
    return false;
  } // createVersionTable()

  /**
   * create aa helper table that will track number of active conections to particular
   * update states od the repository. Because for each previous state that is used we create
   * a separate RO repository. we should track when it is not longer needed
   * (by the number of the requests made for its creation) and remove it permanently
   * @throws SQLException if somthing is fron with the DB connection
   * @return true if table was just created
   */
  protected boolean createActiveStatesTable()
      throws SQLException
  {
    if (!RDBMS().tableExists(ACTIVE_STATES_TABLE)) {
      String tableQ = "CREATE TABLE "+ACTIVE_STATES_TABLE+" ("+
                      "uid " + RDBMS().ID_INT + " NOT NULL,"+
                      "refs " + RDBMS().ID_INT + " )";
      RDBMS().executeUpdate(tableQ);
      return true;
    }
    return false;
  }


  /**
   * it is invoked from super.RemoveExpiredStatements to deal with these id chunks
   * @param idList chunk with ids o fexpired statements
   * @throws SQLException
   */
  protected void _processChunkFromRemoveExpiredStatements(String idList) throws SQLException {
    // the actual ID of the update will be added to UPDATE_TABLE later in _processChangedTriples
    RDBMS().executeUpdate(
      "UPDATE "+ TRIPLES_HIST_TABLE +
      " SET DiedAt = "+ (getStateUid()+1) +
      " WHERE id IN " + idList
      );
  }

/**
 * add's a new explicit statement to repository
 * @param subj is either BNode ot URI denoting the statement's subject
 * @param pred is an URI for predicate
 * @param obj is the object value of the statement (Resource or Literal)