Oracle-OracleConnection

提示：OracleConnection 主要负责与Oracle数据库的交互，特别针对CDC功能，提供了获取和处理数据库更改日志的能力，同时包含数据库连接管理、查询执行和结果处理的通用功能，与DB2Connection作用相似

前言

提示：OracleConnection 类旨在简化与 Oracle 数据库的交互，提供了一套全面的数据库操作接口，特别是针对需要利用 Oracle的变更数据捕获能力的场景。通过这个类，开发者可以更方便地执行数据库操作，而无需直接处理复杂的 JDBC 连接和查询细节。

提示：以下是本篇文章正文内容

一、核心功能

核心功能详细说明

1. 查询字符集 (getNationalCharacterSet 方法):

   • 功能: 查询 Oracle 数据库的 NLS_NCHAR_CHARACTERSET 参数设置。
   • 作用: 确定数据库使用的国家字符集。
   • 与 Debezium 的关联: 这对于确保 Debezium 正确地解析和处理字符串数据至关重要，特别是在处理多语言环境下的文本数据时。

2. 清除日志文件 (removeAllLogFilesFromLogMinerSession 方法):

   • 功能: 从 LogMiner 会话中移除所有已注册的日志文件。
   • 作用: 维护 LogMiner 会话，通过移除已注册的日志文件。
   • 与 Debezium 的关联: 这对于维护 LogMiner 会话和确保 Debezium 能够正确跟踪数据库更改至关重要。LogMiner 是 Oracle 提供的一种机制，用于从重做日志中提取更改记录。

3. 获取重做线程状态 (getRedoThreadState 方法):

   • 功能: 获取 Oracle 数据库的重做线程状态。
   • 作用: 监控重做线程的状态，这对于数据库复制和日志分析非常重要。
   • 与 Debezium 的关联: 这对于监控重做线程的状态，确保 Debezium 能够正确捕捉数据库更改事件至关重要。重做线程状态反映了数据库内部的活动情况，这对于 CDC 功能的稳定性和准确性至关重要。

4. 获取 SQL 关键字 (getSQLKeywords 方法):

   • 功能: 从 JDBC 驱动程序获取支持的 SQL 关键字列表。
   • 作用: 获取数据库支持的关键字列表，帮助构建 SQL 查询时避免语法错误。
   • 与 Debezium 的关联: 这对于构建 SQL 查询时避免语法错误，确保 Debezium 正确地处理 SQL 语句至关重要。这对于处理复杂的数据库结构和优化查询性能很有帮助。

二、代码分析
 

// 定义一个方法，用于将当前会话切换到指定的 PDB
public void setSessionToPdb(String pdbName) {// 声明 Statement 变量，初始值为 nullStatement statement = null;// 尝试块，用于执行可能抛出异常的操作try {// 通过当前连接创建 Statement 对象statement = connection().createStatement();// 执行 SQL 语句，将当前会话的容器设置为指定的 PDB 名称statement.execute("alter session set container=" + pdbName);}// 捕获块，用于处理 SQLException 异常catch (SQLException e) {// 抛出一个新的 RuntimeException，将原始的 SQLException 作为其原因throw new RuntimeException(e);}// finally 块，用于执行无论 try 块是否成功都会执行的操作finally {// 检查 Statement 是否不为 nullif (statement != null) {// 尝试关闭 Statementtry {statement.close();}// 捕获块，用于处理关闭 Statement 时可能发生的 SQLException 异常catch (SQLException e) {// 输出错误日志，记录无法关闭 Statement 的异常LOGGER.error("Couldn't close statement", e);}}}
}

这个方法的作用是将当前会话切换到指定的 PDB。在 Oracle 多租户环境中，一个数据库容器可以包含多个 PDB。通过执行 "alter session set container=" + pdbName 语句，可以将当前会话的上下文切换到指定的 PDB，从而允许后续的数据库操作针对该 PDB 进行。
这种方法对于需要在不同的 PDB 之间切换执行数据库操作的场景非常有用，尤其是在使用 Debezium 这样的工具时，可能需要针对不同的 PDB 捕获变更数据 

// 定义一个方法，用于将当前会话切换回 CDB 的根容器 cdb$root
public void resetSessionToCdb() {// 声明 Statement 变量，初始值为 nullStatement statement = null;// 尝试块，用于执行可能抛出异常的操作try {// 通过当前连接创建 Statement 对象statement = connection().createStatement();// 执行 SQL 语句，将当前会话的容器设置为 cdb$rootstatement.execute("alter session set container=cdb$root");}// 捕获块，用于处理 SQLException 异常catch (SQLException e) {// 抛出一个新的 RuntimeException，将原始的 SQLException 作为其原因throw new RuntimeException(e);}// finally 块，用于执行无论 try 块是否成功都会执行的操作finally {// 检查 Statement 是否不为 nullif (statement != null) {// 尝试关闭 Statementtry {statement.close();}// 捕获块，用于处理关闭 Statement 时可能发生的 SQLException 异常catch (SQLException e) {// 输出错误日志，记录无法关闭 Statement 的异常LOGGER.error("Couldn't close statement", e);}}}
}

方法作用

这个方法的作用是将当前会话切换回 CDB 的根容器 cdb$root。在 Oracle 多租户环境中，一个 CDB 包含一个根容器 cdb$root 和一个或多个可插拔数据库 (PDBs)。通过执行 "alter session set container=cdb$root" 语句，可以将当前会话的上下文切换回 CDB 的根容器，从而允许后续的数据库操作针对整个 CDB 进行。

这种方法对于需要在不同的 PDB 之间切换执行数据库操作后回到 CDB 根容器的场景非常有用，尤其是在使用 Debezium 这样的工具时，可能需要针对整个 CDB 进行某些操作或配置。

     * 获取所有表的TableId集合* * @param catalogName 目录名称* @return 表的TableId集合* @throws SQLException 如果发生数据库异常*/protected Set<TableId> getAllTableIds(String catalogName) throws SQLException {// SQL查询语句，从all_tables表中获取owner和table_name，排除特定的表和索引组织表final String query = "select owner, table_name from all_tables " +"where table_name NOT LIKE 'MDRT_%' " +"and table_name NOT LIKE 'MDRS_%' " +"and table_name NOT LIKE 'MDXT_%' " +"and (table_name NOT LIKE 'SYS_IOT_OVER_%' and IOT_NAME IS NULL) " +"and nested = 'NO'" +"and table_name not in (select PARENT_TABLE_NAME from ALL_NESTED_TABLES)";// 使用HashSet存储查询到的TableIdSet<TableId> tableIds = new HashSet<>();// 执行查询并处理结果集query(query, (rs) -> {while (rs.next()) {// 将查询到的owner和table_name封装成TableId对象，添加到集合中tableIds.add(new TableId(catalogName, rs.getString(1), rs.getString(2)));}// 记录日志，输出查询到的TableIdsLOGGER.trace("TableIds are: {}", tableIds);});// 返回TableId集合return tableIds;}/*** 解析并返回数据库的目录名称* * @param catalogName 目录名称* @return 解析后的目录名称*/@Overrideprotected String resolveCatalogName(String catalogName) {// 从配置中获取pdb名称，如果不存在则使用数据库名称，并转换为大写final String pdbName = config().getString("pdb.name");return (!Strings.isNullOrEmpty(pdbName) ? pdbName : config().getString("dbname")).toUpperCase();}/*** 读取表的唯一索引列表* * @param metadata 数据库元数据* @param id 表的标识符* @return 唯一索引列表* @throws SQLException 如果发生数据库异常*/@Overridepublic List<String> readTableUniqueIndices(DatabaseMetaData metadata, TableId id) throws SQLException {// 调用父类方法，使用双引号包装表标识符return super.readTableUniqueIndices(metadata, id.toDoubleQuoted());}/*** 获取当前时间戳* * @return 当前时间戳* @throws SQLException 如果发生数据库异常*/@Overridepublic Optional<Instant> getCurrentTimestamp() throws SQLException {// 执行SQL查询，返回当前时间戳return queryAndMap("SELECT CURRENT_TIMESTAMP FROM DUAL",rs -> rs.next() ? Optional.of(rs.getTimestamp(1).toInstant()) : Optional.empty());}/*** 判断索引列是否包含在表的唯一索引中* * @param indexName 索引名称* @param columnName 列名称* @return 是否包含在唯一索引中*/@Overrideprotected boolean isTableUniqueIndexIncluded(String indexName, String columnName) {// 如果列名称不为空，且不匹配任何系统列名称模式，则返回trueif (columnName != null) {return !SYS_NC_PATTERN.matcher(columnName).matches()&& !ADT_INDEX_NAMES_PATTERN.matcher(columnName).matches()&& !MROW_PATTERN.matcher(columnName).matches();}// 列名称为空时，返回falsereturn false;}/*** 获取当前的系统更改编号（SCN）* * @return 当前的系统更改编号* @throws SQLException 如果发生数据库异常* @throws IllegalStateException 如果查询未返回至少一行数据*/public Scn getCurrentScn() throws SQLException {// 执行SQL查询并映射结果return queryAndMap("SELECT CURRENT_SCN FROM V$DATABASE", (rs) -> {if (rs.next()) {return Scn.valueOf(rs.getString(1));}// 如果未获取到SCN，抛出异常throw new IllegalStateException("Could not get SCN");});}/*** 生成给定表的DDL元数据* * @param tableId 表标识符，不应为null* @return 生成的DDL* @throws SQLException 如果获取DDL元数据时发生异常* @throws NonRelationalTableException 表不是关系表*/public String getTableMetadataDdl(TableId tableId) throws SQLException, NonRelationalTableException {try {// 查询ALL_ALL_TABLES表，确认表是关系表final String tableType = "SELECT COUNT(1) FROM ALL_ALL_TABLES WHERE OWNER=? AND TABLE_NAME=? AND TABLE_TYPE IS NULL";// 如果查询结果为0，抛出异常if (prepareQueryAndMap(tableType,ps -> {ps.setString(1, tableId.schema());ps.setString(2, tableId.table());},rs -> rs.next() ? rs.getInt(1) : 0) == 0) {throw new NonRelationalTableException("Table " + tableId + " is not a relational table");}// 设置DDL转换参数，排除存储和段属性executeWithoutCommitting("begin dbms_metadata.set_transform_param(DBMS_METADATA.SESSION_TRANSFORM, 'STORAGE', false); end;");executeWithoutCommitting("begin dbms_metadata.set_transform_param(DBMS_METADATA.SESSION_TRANSFORM, 'SEGMENT_ATTRIBUTES', false); end;");// 设置DDL转换参数，启用SQL终止符，以便在返回多个DDL语句时能够分别解析executeWithoutCommitting("begin dbms_metadata.set_transform_param(DBMS_METADATA.SESSION_TRANSFORM, 'SQLTERMINATOR', true); end;");// 执行查询，返回表的DDLreturn prepareQueryAndMap("SELECT dbms_metadata.get_ddl('TABLE',?,?) FROM DUAL",ps -> {ps.setString(1, tableId.table());ps.setString(2, tableId.schema());},rs -> {if (!rs.next()) {throw new DebeziumException("Could not get DDL metadata for table: " + tableId);}Object res = rs.getObject(1);// 返回DDL字符串return ((Clob) res).getSubString(1, (int) ((Clob) res).length());});}finally {// 重置DDL转换参数为默认值executeWithoutCommitting("begin dbms_metadata.set_transform_param(DBMS_METADATA.SESSION_TRANSFORM, 'DEFAULT'); end;");}}

    /*** 检查指定表是否存在。** @param tableId 表标识符，不应为空* @return 如果表存在则返回true，否则返回false* @throws SQLException 如果发生数据库异常*/public boolean isTableExists(TableId tableId) throws SQLException {if (Strings.isNullOrBlank(tableId.schema())) {return prepareQueryAndMap("SELECT COUNT(1) FROM USER_TABLES WHERE TABLE_NAME=?",ps -> ps.setString(1, tableId.table()),rs -> rs.next() && rs.getLong(1) > 0);}return prepareQueryAndMap("SELECT COUNT(1) FROM ALL_TABLES WHERE OWNER=? AND TABLE_NAME=?",ps -> {ps.setString(1, tableId.schema());ps.setString(2, tableId.table());},rs -> rs.next() && rs.getLong(1) > 0);}/*** 判断给定表是否为空。** @param tableId 表标识符，不应为空* @return 如果表没有记录则返回true，否则返回false* @throws SQLException 如果发生数据库异常*/public boolean isTableEmpty(TableId tableId) throws SQLException {return getRowCount(tableId) == 0L;}/*** 获取给定表中的行数。** @param tableId 表标识符，不应为空* @return 表中的行数* @throws SQLException 如果发生数据库异常*/public long getRowCount(TableId tableId) throws SQLException {return queryAndMap("SELECT COUNT(1) FROM " + tableId.toDoubleQuotedString(), rs -> {if (rs.next()) {return rs.getLong(1);}return 0L;});}/*** 执行查询并获取单个可选结果值。** @param <T> 结果类型* @param query 查询语句* @param extractor 结果集提取器* @return 查询结果或null* @throws SQLException 如果发生数据库异常*/public <T> T singleOptionalValue(String query, ResultSetExtractor<T> extractor) throws SQLException {return queryAndMap(query, rs -> rs.next() ? extractor.apply(rs) : null);}/*** 获取归档和重做日志中的第一个系统更改编号（SCN）。** @param archiveLogRetention 归档日志保留时间* @param archiveDestinationName 归档日志目的地名称* @return 最旧的系统更改编号（SCN）* @throws SQLException 如果发生数据库异常* @throws DebeziumException 如果由于没有可用的日志而无法找到最旧的系统更改编号*/public Optional<Scn> getFirstScnInLogs(Duration archiveLogRetention, String archiveDestinationName) throws SQLException {final String oldestFirstChangeQuery = SqlUtils.oldestFirstChangeQuery(archiveLogRetention, archiveDestinationName);final String oldestScn = singleOptionalValue(oldestFirstChangeQuery, rs -> rs.getString(1));if (oldestScn == null) {return Optional.empty();}LOGGER.trace("最旧的SCN在日志中是 '{}'", oldestScn);return Optional.of(Scn.valueOf(oldestScn));}/*** 验证日志位置是否有效。** @param partition 分区信息* @param offset 偏移量上下文* @param config 连接器配置* @return 如果日志位置有效则返回true，否则返回false*/public boolean validateLogPosition(Partition partition, OffsetContext offset, CommonConnectorConfig config) {final Duration archiveLogRetention = ((OracleConnectorConfig) config).getArchiveLogRetention();final String archiveDestinationName = ((OracleConnectorConfig) config).getArchiveLogDestinationName();final Scn storedOffset = ((OracleConnectorConfig) config).getAdapter().getOffsetScn((OracleOffsetContext) offset);try {Optional<Scn> firstAvailableScn = getFirstScnInLogs(archiveLogRetention, archiveDestinationName);return firstAvailableScn.filter(isLessThan(storedOffset)).isPresent();}catch (SQLException e) {throw new DebeziumException("无法获取最新的可用日志位置", e);}}/*** 创建一个判断SCN是否小于存储的SCN的谓词。** @param storedOffset 存储的SCN* @return 谓词*/private static Predicate<Scn> isLessThan(Scn storedOffset) {return scn -> scn.compareTo(storedOffset) < 0;}/*** 构建带有行限制的SQL查询语句。** @param tableId 表标识符* @param limit 查询结果的最大行数* @param projection 投影列* @param condition 条件表达式* @param additionalCondition 额外条件表达式* @param orderBy 排序依据* @return SQL查询字符串*/@Overridepublic String buildSelectWithRowLimits(TableId tableId,int limit,String projection,Optional<String> condition,Optional<String> additionalCondition,String orderBy) {final TableId table = new TableId(null, tableId.schema(), tableId.table());final StringBuilder sql = new StringBuilder("SELECT ");sql.append(projection).append(" FROM ");sql.append(quotedTableIdString(table));if (condition.isPresent()) {sql.append(" WHERE ").append(condition.get());if (additionalCondition.isPresent()) {sql.append(" AND ");sql.append(additionalCondition.get());}}else if (additionalCondition.isPresent()) {sql.append(" WHERE ");sql.append(additionalCondition.get());}if (getOracleVersion().getMajor() < 12) {sql.insert(0, " SELECT * FROM (").append(" ORDER BY ").append(orderBy).append(")").append(" WHERE ROWNUM <=").append(limit);}else {sql.append(" ORDER BY ").append(orderBy).append(" FETCH NEXT ").append(limit).append(" ROWS ONLY");}return sql.toString();}

    /*** 检查数据库是否处于归档日志模式。* * @return 如果数据库处于归档日志模式，则返回true；否则返回false。*/protected boolean isArchiveLogMode() {try {final String mode = queryAndMap("SELECT LOG_MODE FROM V$DATABASE", rs -> rs.next() ? rs.getString(1) : "");LOGGER.debug("LOG_MODE={}", mode);return "ARCHIVELOG".equalsIgnoreCase(mode);}catch (SQLException e) {throw new DebeziumException("Unexpected error while connecting to Oracle and looking at LOG_MODE mode: ", e);}}/*** 将系统改变编号（SCN）解析为时间戳，返回值处于数据库时区。* * SCN到时间戳的映射仅在闪回查询区域期间保留。这意味着最终这些值之间的映射不再由Oracle保持，* 使用一个已经过期的SCN值进行调用将导致ORA-08181错误。此函数显式检查此用例，如果抛出ORA-08181错误，* 则被视为不存在该值，返回一个空的可选值。* * @param scn 系统改变编号，不得为null* @return 一个可选的时间戳，表示系统改变编号发生的时间* @throws SQLException 如果发生数据库异常*/public Optional<Instant> getScnToTimestamp(Scn scn) throws SQLException {try {return queryAndMap("SELECT scn_to_timestamp('" + scn + "') FROM DUAL", rs -> rs.next()? Optional.of(rs.getTimestamp(1).toInstant()): Optional.empty());}catch (SQLException e) {if (e.getMessage().startsWith("ORA-08181")) {// ORA-08181 specified number is not a valid system change number// This happens when the SCN provided is outside the flashback area range// This should be treated as a value is not available rather than an errorreturn Optional.empty();}// Any other SQLException should be thrownthrow e;}}/*** 根据时间调整SCN。* * @param scn 原始SCN值* @param adjustment 要应用的时间调整（正或负）* @return 调整后的SCN值* @throws SQLException 如果发生数据库异常且无法计算调整后的SCN*/public Scn getScnAdjustedByTime(Scn scn, Duration adjustment) throws SQLException {try {final String result = prepareQueryAndMap("SELECT timestamp_to_scn(scn_to_timestamp(?) - (? / 86400000)) FROM DUAL",st -> {st.setString(1, scn.toString());st.setLong(2, adjustment.toMillis());},singleResultMapper(rs -> rs.getString(1), "Failed to get adjusted SCN from: " + scn));return Scn.valueOf(result);}catch (SQLException e) {if (e.getErrorCode() == 8181 || e.getErrorCode() == 8180) {// This happens when the SCN provided is outside the flashback/undo areareturn Scn.NULL;}throw e;}}/*** 检查指定的归档日志目标是否有效。* * @param archiveDestinationName 归档日志目标名称* @return 如果目标有效返回true，否则返回false* @throws SQLException 如果无法连接到数据库或目标名称无效*/public boolean isArchiveLogDestinationValid(String archiveDestinationName) throws SQLException {return prepareQueryAndMap("SELECT STATUS, TYPE FROM V$ARCHIVE_DEST_STATUS WHERE DEST_NAME=?",st -> st.setString(1, archiveDestinationName),rs -> {if (!rs.next()) {throw new DebeziumException(String.format("Archive log destination name '%s' is unknown to Oracle",archiveDestinationName));}return "VALID".equals(rs.getString("STATUS")) && "LOCAL".equals(rs.getString("TYPE"));});}/*** 检查是否只有一个归档日志目标有效。* * @return 如果只有一个归档日志目标有效，则返回true；否则返回false。* @throws SQLException 如果无法确定归档日志目标的数量*/public boolean isOnlyOneArchiveLogDestinationValid() throws SQLException {return queryAndMap("SELECT COUNT(1) FROM V$ARCHIVE_DEST_STATUS WHERE STATUS='VALID' AND TYPE='LOCAL'",rs -> {if (!rs.next()) {throw new DebeziumException("Unable to resolve number of archive log destinations");}return rs.getLong(1) == 1L;});}/*** 重写列编辑器，以在解析默认值之前调整列状态。* * 这允许在解析默认值之前覆盖列状态，从而使默认值的输出与列值具有相同的精度。* * @param column 待重写的列编辑器* @return 调整后的列编辑器*/@Overrideprotected ColumnEditor overrideColumn(ColumnEditor column) {// This allows the column state to be overridden before default-value resolution so that the// output of the default value is within the same precision as that of the column values.if (OracleTypes.TIMESTAMP == column.jdbcType()) {column.length(column.scale().orElse(Column.UNSET_INT_VALUE)).scale(null);}else if (OracleTypes.NUMBER == column.jdbcType()) {column.scale().filter(s -> s == ORACLE_UNSET_SCALE).ifPresent(s -> column.scale(null));}return column;}

     * 按需懒查询并缓存该值。** <a href="https://docs.oracle.com/en/database/oracle/oracle-database/19/sqlrf/Data-Types.html#GUID-FE15E51B-52C6-45D7-9883-4DF47716A17D">NCHAR</a>* <a href="https://docs.oracle.com/en/database/oracle/oracle-database/19/sqlrf/Data-Types.html#GUID-CC15FC97-BE94-4FA4-994A-6DDF7F1A9904">NVARCHAR2</a>** @return 字符集，只能是 {@code AL16UTF16} 或 {@code UTF8}。*/public CharacterSet getNationalCharacterSet() {final String query = "select VALUE from NLS_DATABASE_PARAMETERS where PARAMETER = 'NLS_NCHAR_CHARACTERSET'";try {final String nlsCharacterSet = queryAndMap(query, rs -> {if (rs.next()) {return rs.getString(1);}return null;});if (nlsCharacterSet != null) {switch (nlsCharacterSet) {case "AL16UTF16":return CharacterSet.make(CharacterSet.AL16UTF16_CHARSET);case "UTF8":return CharacterSet.make(CharacterSet.UTF8_CHARSET);}}throw new SQLException("检测到意外的 NLS_NCHAR_CHARACTERSET: " + nlsCharacterSet);}catch (SQLException e) {throw new DebeziumException("无法解析 Oracle 的 NLS_NCHAR_CHARACTERSET 属性", e);}}public void removeAllLogFilesFromLogMinerSession() throws SQLException {final Set<String> fileNames = queryAndMap("SELECT FILENAME AS NAME FROM V$LOGMNR_LOGS", rs -> {final Set<String> results = new HashSet<>();while (rs.next()) {results.add(rs.getString(1));}return results;});for (String fileName : fileNames) {LOGGER.debug("从 LogMiner 会话中移除文件 {}。", fileName);final String sql = "BEGIN SYS.DBMS_LOGMNR.REMOVE_LOGFILE(LOGFILENAME => '" + fileName + "');END;";try (CallableStatement statement = connection(false).prepareCall(sql)) {statement.execute();}}}public RedoThreadState getRedoThreadState() throws SQLException {final String query = "SELECT * FROM V$THREAD";try {return queryAndMap(query, rs -> {RedoThreadState.Builder builder = RedoThreadState.builder();while (rs.next()) {// 尽管这个字段实际上不应该为 NULL，但数据库元数据允许这样做final int threadId = rs.getInt("THREAD#");if (!rs.wasNull()) {RedoThreadState.RedoThread.Builder threadBuilder = builder.thread().threadId(threadId).status(rs.getString("STATUS")).enabled(rs.getString("ENABLED")).logGroups(rs.getLong("GROUPS")).instanceName(rs.getString("INSTANCE")).openTime(readTimestampAsInstant(rs, "OPEN_TIME")).currentGroupNumber(rs.getLong("CURRENT_GROUP#")).currentSequenceNumber(rs.getLong("SEQUENCE#")).checkpointScn(readScnColumnAsScn(rs, "CHECKPOINT_CHANGE#")).checkpointTime(readTimestampAsInstant(rs, "CHECKPOINT_TIME")).enabledScn(readScnColumnAsScn(rs, "ENABLE_CHANGE#")).enabledTime(readTimestampAsInstant(rs, "ENABLE_TIME")).disabledScn(readScnColumnAsScn(rs, "DISABLE_CHANGE#")).disabledTime(readTimestampAsInstant(rs, "DISABLE_TIME"));if (getOracleVersion().getMajor() >= 11) {threadBuilder = threadBuilder.lastRedoSequenceNumber(rs.getLong("LAST_REDO_SEQUENCE#")).lastRedoBlock(rs.getLong("LAST_REDO_BLOCK")).lastRedoScn(readScnColumnAsScn(rs, "LAST_REDO_CHANGE#")).lastRedoTime(readTimestampAsInstant(rs, "LAST_REDO_TIME"));}if (getOracleVersion().getMajor() >= 12) {threadBuilder = threadBuilder.conId(rs.getLong("CON_ID"));}builder = threadBuilder.build();}}return builder.build();});}catch (SQLException e) {throw new DebeziumException("无法读取 Oracle 数据库的重做线程状态", e);}}public List<String> getSQLKeywords() {try {return Arrays.asList(connection().getMetaData().getSQLKeywords().split(","));}catch (SQLException e) {LOGGER.debug("无法从 JDBC 驱动程序获取 SQL 关键字。", e);return Collections.emptyList();}}private static Scn readScnColumnAsScn(ResultSet rs, String columnName) throws SQLException {final String value = rs.getString(columnName);return Strings.isNullOrEmpty(value) ? Scn.NULL : Scn.valueOf(value);}private static Instant readTimestampAsInstant(ResultSet rs, String columnName) throws SQLException {final Timestamp value = rs.getTimestamp(columnName);return value == null ? null : value.toInstant();}

总结

1. getNationalCharacterSet:

   • 查询并返回数据库的 NLS_NCHAR_CHARACTERSET 设置。
   • 返回值只能是 AL16UTF16 或 UTF8。
   • 如果查询结果不是预期中的字符集，则抛出异常。

2. removeAllLogFilesFromLogMinerSession:

   • 从 LogMiner 会话中移除所有日志文件。
   • 首先查询所有日志文件名。
   • 遍历这些文件名，并执行 PL/SQL 块来移除每个文件。

3. getRedoThreadState:

   • 获取重做线程的状态信息。
   • 查询 V$THREAD 表以获取重做线程详情。
   • 构建并返回一个表示重做线程状态的对象。

4. getSQLKeywords:

   • 通过 JDBC 获取数据库支持的所有 SQL 关键字。
   • 返回关键字列表。

5. 辅助方法：

   • readScnColumnAsScn: 将结果集中特定列的字符串值转换为 Scn 对象。
   • readTimestampAsInstant: 将结果集中的 Timestamp 转换为 Instant。
   • 定义了两个函数式接口 ContainerWork 和 ObjectIdentifierConsumer 供其他代码使用。