public static String build(OracleConnectorConfig connectorConfig, OracleDatabaseSchema schema) {
        final StringBuilder query = new StringBuilder(1024);
        query.append("SELECT SCN, SQL_REDO, OPERATION_CODE, TIMESTAMP, XID, CSF, TABLE_NAME, SEG_OWNER, OPERATION, ");
        query.append("USERNAME, ROW_ID, ROLLBACK, RS_ID ");
        query.append("FROM ").append(LOGMNR_CONTENTS_VIEW).append(" ");

        // These bind parameters will be bound when the query is executed by the caller.
        query.append("WHERE SCN > ? AND SCN <= ? ");

        // Restrict to configured PDB if one is supplied
        final String pdbName = connectorConfig.getPdbName();
        if (!Strings.isNullOrEmpty(pdbName)) {
            query.append("AND ").append("SRC_CON_NAME = '").append(pdbName.toUpperCase()).append("' ");
        }

        query.append("AND (");

        // Always include START, COMMIT, MISSING_SCN, and ROLLBACK operations
        query.append("(OPERATION_CODE IN (6,7,34,36)");

        if (!schema.storeOnlyCapturedTables()) {
            // In this mode, the connector will always be fed DDL operations for all tables even if they
            // are not part of the inclusion/exclusion lists.
            query.append(" OR ").append(buildDdlPredicate()).append(" ");
            // Insert, Update, Delete, SelectLob, LobWrite, LobTrim, and LobErase
            if (connectorConfig.isLobEnabled()) {
                query.append(") OR (OPERATION_CODE IN (1,2,3,9,10,11,29) ");
            } else {
                query.append(") OR (OPERATION_CODE IN (1,2,3) ");
            }
        } else {
            // Insert, Update, Delete, SelectLob, LobWrite, LobTrim, and LobErase
            if (connectorConfig.isLobEnabled()) {
                query.append(") OR ((OPERATION_CODE IN (1,2,3,9,10,11,29) ");
            } else {
                query.append(") OR ((OPERATION_CODE IN (1,2,3) ");
            }
            // In this mode, the connector will filter DDL operations based on the table inclusion/exclusion lists
            query.append("OR ").append(buildDdlPredicate()).append(") ");
        }

        // Always ignore the flush table
        query.append("AND TABLE_NAME != '").append(SqlUtils.LOGMNR_FLUSH_TABLE).append("' ");

        // There are some common schemas that we automatically ignore when building the runtime Filter
        // predicates and we put that same list of schemas here and apply those in the generated SQL.
        if (!OracleConnectorConfig.EXCLUDED_SCHEMAS.isEmpty()) {
            query.append("AND SEG_OWNER NOT IN (");
            for (Iterator<String> i = OracleConnectorConfig.EXCLUDED_SCHEMAS.iterator(); i.hasNext();) {
                String excludedSchema = i.next();
                query.append("'").append(excludedSchema.toUpperCase()).append("'");
                if (i.hasNext()) {
                    query.append(",");
                }
            }
            query.append(") ");
        }

        String schemaPredicate = buildSchemaPredicate(connectorConfig);
        if (!Strings.isNullOrEmpty(schemaPredicate)) {
            query.append("AND ").append(schemaPredicate).append(" ");
        }

        String tablePredicate = buildTablePredicate(connectorConfig);
        if (!Strings.isNullOrEmpty(tablePredicate)) {
            query.append("AND ").append(tablePredicate).append(" ");
        }

        query.append("))");

        return query.toString();
    }
Builds the LogMiner contents view query.

The returned query will contain 2 bind parameters that the caller is responsible for binding before
executing the query.  The first bind parameter is the lower-bounds of the SCN mining window that is
not-inclusive while the second is the upper-bounds of the SCN mining window that is inclusive.

The built query relies on the following columns from V$LOGMNR_CONTENTS:
<pre>
    SCN - the system change number at which the change was made
    SQL_REDO - the reconstructed SQL statement that initiated the change
    OPERATION - the database operation type name
    OPERATION_CODE - the database operation numeric code
    TIMESTAMP - the time when the change event occurred
    XID - the transaction identifier the change participated in
    CSF - the continuation flag, identifies rows that should be processed together as single row, 0=no, 1=yes
    TABLE_NAME - the name of the table for which the change is for
    SEG_OWNER - the name of the schema for which the change is for
    USERNAME - the name of the database user that caused the change
    ROW_ID - the unique identifier of the row that the change is for, may not always be set with valid value
    ROLLBACK - the rollback flag, value of 0 or 1.  1 implies the row was rolled back
    RS_ID - the rollback segment idenifier where the change record was record from
</pre>

@param connectorConfig connector configuration, should not be {@code null}
@param schema database schema, should not be {@code null}
@return the SQL string to be used to fetch changes from Oracle LogMiner
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/io/debezium/connector/oracle/logminer/LogMinerQueryBuilder.java