神通数据库的Kettle数据库插件

package org.pentaho.di.core.database; import java.sql.ResultSet; import org.pentaho.di.core.Const; import org.pentaho.di.core.exception.KettleDatabaseException; import org.pentaho.di.core.plugins.DatabaseMetaPlugin; import org.pentaho.di.core.row.ValueMetaInterface; import org.pentaho.di.core.util.Utils; /** * DatabaseMeta数据库插件-神通数据库 */ @DatabaseMetaPlugin(type = "OSCAR", typeDescription = "神通数据库") public class OscarDatabaseMeta extends BaseDatabaseMeta implements DatabaseInterface { private static final String STRICT_BIGNUMBER_INTERPRETATION = "STRICT_NUMBER_38_INTERPRETATION"; @Override public int[] getAccessTypeList() { return new int[] { DatabaseMeta.TYPE_ACCESS_NATIVE, DatabaseMeta.TYPE_ACCESS_JNDI }; } @Override public int getDefaultDatabasePort() { if (getAccessType() == DatabaseMeta.TYPE_ACCESS_NATIVE) { return 2003; } return -1; } /** * 当前数据库是否支持自增类型的字段 */ @Override public boolean supportsAutoInc() { return false; } /** * 获取限制读取条数的数据，追加再select语句后实现限制返回的结果数 * @see org.pentaho.di.core.database.DatabaseInterface#getLimitClause(int) */ @Override public String getLimitClause(int nrRows) { return " WHERE ROWNUM <= " + nrRows; } /** * 返回获取表所有字段信息的语句 * @param tableName * @return The SQL to launch. */ @Override public String getSQLQueryFields(String tableName) { return "SELECT * FROM " + tableName + " WHERE 1=0"; } @Override public String getSQLTableExists(String tablename) { return getSQLQueryFields(tablename); } @Override public String getSQLColumnExists(String columnname, String tablename) { return getSQLQueryColumnFields(columnname, tablename); } public String getSQLQueryColumnFields(String columnname, String tableName) { return "SELECT " + columnname + " FROM " + tableName + " WHERE 1=0"; } @Override public boolean needsToLockAllTables() { return false; } @Override public String getDriverClass() { if (getAccessType() == DatabaseMeta.TYPE_ACCESS_ODBC) { return "sun.jdbc.odbc.JdbcOdbcDriver"; } else { return "com.oscar.Driver"; } } @Override public String getURL(String hostname, String port, String databaseName) throws KettleDatabaseException { if (getAccessType() == DatabaseMeta.TYPE_ACCESS_ODBC) { return "jdbc:odbc:" + databaseName; } else if (getAccessType() == DatabaseMeta.TYPE_ACCESS_NATIVE) { // <host>/<database> // <host>:<port>/<database> String _hostname = hostname; String _port = port; String _databaseName = databaseName; if (Utils.isEmpty(hostname)) { _hostname = "localhost"; } if (Utils.isEmpty(port) || port.equals("-1")) { _port = ""; } if (Utils.isEmpty(databaseName)) { throw new KettleDatabaseException("必须指定数据库名称"); } if (!databaseName.startsWith("/")) { _databaseName = "/" + databaseName; } return "jdbc:oscar://" + _hostname + (Utils.isEmpty(_port) ? "" : ":" + _port) + _databaseName; } else { throw new KettleDatabaseException("不支持的数据库连接方式[" + getAccessType() + "]"); } } /** * Oracle doesn't support options in the URL, we need to put these in a * Properties object at connection time... */ @Override public boolean supportsOptionsInURL() { return false; } /** * @return true if the database supports sequences */ @Override public boolean supportsSequences() { return true; } /** * Check if a sequence exists. * * @param sequenceName * The sequence to check * @return The SQL to get the name of the sequence back from the databases data * dictionary */ @Override public String getSQLSequenceExists(String sequenceName) { int dotPos = sequenceName.indexOf('.'); String sql = ""; if (dotPos == -1) { // if schema is not specified try to get sequence which belongs to current user sql = "SELECT * FROM USER_SEQUENCES WHERE SEQUENCE_NAME = '" + sequenceName.toUpperCase() + "'"; } else { String schemaName = sequenceName.substring(0, dotPos); String seqName = sequenceName.substring(dotPos + 1); sql = "SELECT * FROM ALL_SEQUENCES WHERE SEQUENCE_NAME = '" + seqName.toUpperCase() + "' AND SEQUENCE_OWNER = '" + schemaName.toUpperCase() + "'"; } return sql; } /** * Get the current value of a database sequence * * @param sequenceName * The sequence to check * @return The current value of a database sequence */ @Override public String getSQLCurrentSequenceValue(String sequenceName) { return "SELECT " + sequenceName + ".currval FROM DUAL"; } /** * Get the SQL to get the next value of a sequence. (Oracle only) * * @param sequenceName * The sequence name * @return the SQL to get the next value of a sequence. (Oracle only) */ @Override public String getSQLNextSequenceValue(String sequenceName) { return "SELECT " + sequenceName + ".nextval FROM dual"; } @Override public boolean supportsSequenceNoMaxValueOption() { return true; } /** * @return true if we need to supply the schema-name to getTables in order to * get a correct list of items. */ @Override public boolean useSchemaNameForTableList() { return true; } /** * @return true if the database supports synonyms */ @Override public boolean supportsSynonyms() { return true; } /** * Generates the SQL statement to add a column to the specified table * * @param tablename * The table to add * @param v * The column defined as a value * @param tk * the name of the technical key field * @param use_autoinc * whether or not this field uses auto increment * @param pk * the name of the primary key field * @param semicolon * whether or not to add a semi-colon behind the statement. * @return the SQL statement to add a column to the specified table */ @Override public String getAddColumnStatement(String tablename, ValueMetaInterface v, String tk, boolean use_autoinc, String pk, boolean semicolon) { return "ALTER TABLE " + tablename + " ADD " + getFieldDefinition(v, tk, pk, use_autoinc, true, false); } /** * Generates the SQL statement to drop a column from the specified table * * @param tablename * The table to add * @param v * The column defined as a value * @param tk * the name of the technical key field * @param use_autoinc * whether or not this field uses auto increment * @param pk * the name of the primary key field * @param semicolon * whether or not to add a semi-colon behind the statement. * @return the SQL statement to drop a column from the specified table */ @Override public String getDropColumnStatement(String tablename, ValueMetaInterface v, String tk, boolean use_autoinc, String pk, boolean semicolon) { return "ALTER TABLE " + tablename + " DROP COLUMN " + v.getName() + Const.CR; } /** * Generates the SQL statement to modify a column in the specified table * * @param tablename * The table to add * @param v * The column defined as a value * @param tk * the name of the technical key field * @param use_autoinc * whether or not this field uses auto increment * @param pk * the name of the primary key field * @param semicolon * whether or not to add a semi-colon behind the statement. * @return the SQL statement to modify a column in the specified table */ @Override public String getModifyColumnStatem