Перейти к содержимому
Последнее обновление

Руководство разработчика по добавлению источников данных в JDBC Catalog

Этот документ предназначен для разработчиков ядра, которым нужно добавить новый источник данных в JDBC Catalog Apache Doris. В нём описаны классы, которые необходимо изменить на уровнях FE и BE, сигнатуры ключевых методов и правила написания регрессионных тестов. В приведённых ниже примерах добавляется новый источник данных с именем NewDB.

Применимая версия: Apache Doris 3.0 и более поздние версии.

Оглавление​

1. Общая архитектура​

JdbcCatalog использует архитектуру с разделением на Frontend (FE) и Backend (BE):

УровеньОбязанности
Frontend (FE)Управление метаданными, разбор SQL, планирование и оптимизация запросов
Backend (BE)Сканирование данных, преобразование типов и выполнение

Основные компоненты FE​

КомпонентРоль
JdbcResourceОпределяет ресурс подключения JDBC и его параметры
JdbcExternalCatalogУправляет всем Catalog, создаёт JdbcClient и управляет им
JdbcExternalDatabaseУправляет метаданными на уровне базы данных
JdbcExternalTableУправляет метаданными и схемой на уровне таблицы
JdbcClientАбстрактный базовый класс для операций с метаданными (список баз данных, список таблиц, сведения о столбцах и т. д.)
JdbcScanNodeУзел сканирования в плане запроса, отвечающий за формирование SQL-запроса
JdbcTableSinkУзел приёма данных в плане записи

Основные компоненты BE​

КомпонентРоль
vjdbc_connector.cppКоннектор на C++, вызывающий исполнитель Java через JNI. Разработчикам не нужно изменять этот файл.
BaseJdbcExecutorАбстрактный базовый класс исполнителя Java; обрабатывает операции чтения и записи данных
JdbcExecutorFactoryФабричный класс, создающий соответствующий исполнитель в зависимости от типа источника данных

2. Обзор этапов разработки​

ШагЦельЗатрагиваемые файлы
Шаг 1Зарегистрировать новый тип источника данныхgensrc/thrift/Types.thrift
Шаг 2Обеспечить распознавание URL и типа на стороне FEJdbcResource.java, JdbcTable.java
Шаг 3Реализовать клиент метаданныхJdbcNewDBClient.java, JdbcClient.java (регистрация в фабрике)
Шаг 4Адаптировать план запросаJdbcScanNode.java
Шаг 5Реализовать исполнитель операций с данными на стороне BENewDBJdbcExecutor.java, JdbcExecutorFactory.java
Шаг 6Добавить регрессионные тестыregression-test/suites/external_table_p0/jdbc/...

3. Шаг 1 - Обновление определения Thrift​

  1. Измените файл Thrift: В gensrc/thrift/Types.thrift добавьте новый тип в перечисление TOdbcTableType.

  2. Сгенерируйте код: Запустите скрипт в корневом каталоге Doris, чтобы новое значение перечисления стало доступно в коде Java и C++.

4. Шаг 2 - Определение основных метаданных в FE​

  1. Измените JdbcResource.java: Добавьте константы для префикса URL и имени типа NewDB, а также логику распознавания в метод parseDbType.

  2. Измените JdbcTable.java: Сопоставьте строку типа NewDB со значением перечисления Thrift и определите правила заключения SQL-идентификаторов в кавычки (например, символ кавычек для идентификаторов).

5. Шаг 3 - Реализация клиента метаданных FE JdbcClient​

5.1 Основная логика работы с метаданными​

На этом шаге реализуется основная логика взаимодействия с метаданными внешнего источника данных. Создайте JdbcNewDBClient.java в каталоге doris/fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/ и переопределите следующие ключевые методы:

// doris/fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcNewDBClient.java

public class JdbcNewDBClient extends JdbcClient {

    public JdbcNewDBClient(JdbcClientConfig jdbcClientConfig) {
        super(jdbcClientConfig);
    }

    /**
     * [Must override] Get the list of databases (or schemas).
     *
     * @return The list of database names.
     * @purpose This is the underlying implementation of the `SHOW DATABASES` command.
     * @implementation
     * 1. Get a JDBC Connection from the connection pool.
     * 2. Use `connection.getMetaData().getCatalogs()` or `getSchemas()` to get the list.
     * 3. Iterate the ResultSet and add the database names to a List<String>.
     * 4. Call `filterDatabaseNames()` to filter out system databases and databases that do not match the rules.
     */
    @Override
    public List<String> getDatabaseNameList() {
        // Implement the logic to get all database names of NewDB
    }

    /**
     * This is the underlying implementation of the `SHOW TABLES` command.
     *
     * @param remoteDbName The database name.
     * @return The list of table names.
     */
    @Override
    public List<String> getTablesNameList(String remoteDbName) {
        // Usually you do not need to override this method directly; override the internally called processTable() instead
        return super.getTablesNameList(remoteDbName);
    }

    /**
     * [Must override] Get the metadata of a table.
     *
     * @purpose Called by methods such as getTablesNameList() and isTableExist()
     *          to perform metadata lookup through `DatabaseMetaData.getTables()`.
     * @implementation
     * 1. Get the JDBC Connection and DatabaseMetaData.
     * 2. Call `databaseMetaData.getTables(catalog, schemaPattern, tableNamePattern, types)`.
     * 3. The key is to pass the correct values for the `catalog` and `schemaPattern` parameters, which depends on the JDBC driver implementation of NewDB.
     *    - If NewDB uses Catalog, pass `remoteDbName` as catalog and `null` as schemaPattern.
     *    - If NewDB uses Schema, pass `null` as catalog and `remoteDbName` as schemaPattern.
     */
    @Override
    protected void processTable(String remoteDbName, String remoteTableName, String[] tableTypes,
            Consumer<ResultSet> resultSetConsumer) {
        // Implement the logic that calls DatabaseMetaData.getTables()
    }

    /**
     * [Must override] **Core method** - Define the mapping from NewDB types to Doris types.
     *
     * @param fieldSchema Contains the column information obtained from the JDBC Driver, such as type name (getDataTypeName),
     *                    precision (getColumnSize), scale (getDecimalDigits), and so on.
     * @return The corresponding Doris `Type`.
     */
    @Override
    public Type jdbcTypeToDoris(JdbcFieldSchema fieldSchema) {
        // Return the Doris Type based on the information in fieldSchema
        // For example:
        // String newdbType = fieldSchema.getDataTypeName().toUpperCase();
        // switch (newdbType) {
        //     case "VARCHAR": return ScalarType.createStringType();
        //     case "INTEGER": return Type.INT;
        //     case "DECIMAL":
        //         return ScalarType.createDecimalV3Type(fieldSchema.getColumnSize(), fieldSchema.getDecimalDigits());
        //     // ... other type mappings
        //     default: return Type.UNSUPPORTED;
        // }
    }

    /**
     * [Recommended override] Define the system databases that should be filtered out from the `getDatabaseNameList` result.
     */
    @Override
    protected Set<String> getFilterInternalDatabases() {
        // return ImmutableSet.of("information_schema", "sys", "performance_schema", "newdb_system_db");
    }

    /**
     * [Recommended override] Define the query statement used when testing the connection.
     * The default is "SELECT 1", but some databases (such as Oracle) require "SELECT 1 FROM DUAL".
     */
    @Override
    public String getTestQuery() {
        // return "SELECT 1"; // Or a simple query supported by NewDB
    }

    /**
     * [Optional override] Define what value the catalog parameter should take when calling `DatabaseMetaData`-related methods.
     * The default returns `connection.getCatalog()`. If NewDB does not support Catalog, return `null`.
     */
    @Override
    protected String getCatalogName(Connection conn) throws SQLException {
        // return conn.getCatalog(); // Or return null;
    }

    /**
     * [Optional override] Get column definitions.
     * The default implementation is usually enough, but for some databases (such as MySQL), you may need to execute a SHOW or SELECT query
     * to get more precise type information (for example, to distinguish `TINYINT(1)` from `TINYINT`).
     */
    @Override
    public List<JdbcFieldSchema> getJdbcColumnsInfo(String remoteDbName, String remoteTableName) {
        return super.getJdbcColumnsInfo(remoteDbName, remoteTableName);
    }
}

5.2 Регистрация JdbcNewDBClient в фабричном классе​

Измените doris/fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/client/JdbcClient.java так, чтобы фабричный метод возвращал экземпляр JdbcNewDBClient в зависимости от TOdbcTableType.

6. Шаг 4 - Адаптация плана запроса FE JdbcScanNode​

В методе JdbcScanNode.getJdbcQueryStr() добавьте логику формирования предложения LIMIT для NewDB. В примере ниже показаны три типичные ветви: MSSQL (TOP n), стандартный LIMIT и FETCH FIRST n ROWS ONLY:

// doris/fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/source/JdbcScanNode.java

private String getJdbcQueryStr() {
    // ...
    // MSSQL use select top to do top n
    if (shouldPushDownLimit() && jdbcType == TOdbcTableType.SQLSERVER) {
        sql.append("TOP " + limit + " ");
    }
    // ...
    sql.append(Joiner.on(", ").join(columns));
    sql.append(" FROM ").append(tableName);
    // ...
    // Other DataBase use limit do top n
    if (shouldPushDownLimit()
            && (jdbcType == TOdbcTableType.MYSQL
            || jdbcType == TOdbcTableType.POSTGRESQL
            /* Other databases that use the standard LIMIT */
            // If NewDB also uses the standard LIMIT, add it here
            )) {
        sql.append(" LIMIT ").append(limit);
    }
    if (shouldPushDownLimit() && jdbcType == TOdbcTableType.NEWDB) {
        // Example: if NewDB has a special LIMIT syntax such as 'FETCH FIRST n ROWS ONLY',
        // handle it separately here
        sql.append(" FETCH FIRST ").append(limit).append(" ROWS ONLY");
    }
    return sql.toString();
}

7. Шаг 5 - Реализация исполнителя операций с данными BE JdbcExecutor​

7.1 Основная логика чтения данных​

На этом шаге реализуется непосредственное чтение и запись данных через JDBC на стороне BE с использованием JNI. Создайте NewDBJdbcExecutor.java в каталоге doris/fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/ и переопределите следующие ключевые методы:

// doris/fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/NewDBJdbcExecutor.java

public class NewDBJdbcExecutor extends BaseJdbcExecutor {

    public NewDBJdbcExecutor(byte[] thriftParams) throws Exception {
        super(thriftParams);
    }

    /**
     * [Override as needed] Initialize the Java-side cache that holds a batch of data.
     *
     * @purpose This is a key performance optimization. It pre-allocates memory for the entire batch before iterating the ResultSet.
     *          This avoids creating objects repeatedly in the loop and significantly improves data read performance.
     * @implementation
     * 1. Iterate over each column.
     * 2. For most standard types, the best practice is to call `outputTable.getColumn(i).newObjectContainerArray(batchSizeNum)`.
     *    This creates a type-safe array container (such as Integer[], BigDecimal[], Long[], etc.) based on the Doris target type.
     * 3. For types that need special handling (for example, a type returned as a binary object by the JDBC Driver but that you only want to handle as a String),
     *    allocate a more generic container, such as `Object[]`.
     */
    @Override
    protected void initializeBlock(int columnCount, String[] replaceStringList, int batchSizeNum,
            VectorTable outputTable) {
        for (int i = 0; i < columnCount; ++i) {
            if (outputTable.getColumnType(i).getType() == Type.STRING) {
                block.add(new Object[batchSizeNum]);
            } else {
                block.add(outputTable.getColumn(i).newObjectContainerArray(batchSizeNum));
            }
        }
    }

    /**
     * [Must override] **Core method** - Get the data of a single column from the JDBC ResultSet.
     *
     * @param columnIndex The column index (starting from 0).
     * @param type The column type expected on the Doris side.
     * @return The Java object fetched and converted from the ResultSet.
     * @implementation
     * 1. Use `resultSet.getObject(columnIndex + 1, TargetClass.class)` to get the value; this is the safest way.
     * 2. `TargetClass.class` should match the Doris `type`. For example, `Type.INT` corresponds to `Integer.class`.
     * 3. You must handle `NULL` values correctly via `resultSet.wasNull()`.
     */
    @Override
    protected Object getColumnValue(int columnIndex, ColumnType type, String[] replaceStringList) throws SQLException {
        switch (type.getType()) {
            case INT:
                return resultSet.getObject(columnIndex + 1, Integer.class);
            case DATETIME:
                return resultSet.getObject(columnIndex + 1, LocalDateTime.class);
            // ... other types
            default:
                 return resultSet.getObject(columnIndex + 1);
        }
    }

    /**
     * [Must override] Provide an output converter for specific types.
     *
     * @purpose Do final processing on the value returned by `getColumnValue` before passing it to the C++ layer.
     *          Commonly used in scenarios that require specific formatting.
     * @implementation
     *   A common example is the `java.sql.Time` type. To preserve its microsecond precision, `getColumnValue`
     *   gets a `Time` object, and the converter formats it into the `HH:mm:ss.SSSSSS` string that Doris requires.
     *   Another example is converting a `byte[]` to a hex-display string.
     */
    @Override
    protected ColumnValueConverter getOutputConverter(ColumnType columnType, String replaceString) {
        if (columnType.getType() == ColumnType.Type.STRING) {
            // Example: how the MySQL Executor handles the Time type
            return createConverter(input -> {
                if (input instanceof java.sql.Time) {
                    return timeToString((java.sql.Time) input); // timeToString is a method in BaseJdbcExecutor
                }
                return input.toString();
            }, String.class);
        }
        return null; // Other types usually do not need conversion
    }

    /**
     * [Optional override] Initialize the PreparedStatement.
     *
     * @purpose Use this when you need to optimize the size and method of certain write or read operations.
     * @implementation
     * 1. You must handle read (READ) and write (WRITE) operations separately.
     * 2. For read operations, use `stmt.setFetchSize()` to control how many rows the JDBC Driver pulls from the database at one time.
     *    This avoids OOM. Different drivers behave differently for this parameter (for example, setting it to `Integer.MIN_VALUE` in MySQL enables streaming reads).
     */
    @Override
    protected void initializeStatement(Connection conn, JdbcDataSourceConfig config, String sql) throws SQLException {
        // if (config.getOp() == TJdbcOperation.READ) {
        //     stmt = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
        //     stmt.setFetchSize(1024); // Set an appropriate fetch size for NewDB
        // } else {
        //     preparedStatement = conn.prepareStatement(sql);
        // }
    }

    /**
     * [Optional override] Interrupt the JDBC connection when the query is canceled.
     *
     * @purpose Override this if the driver of a data source has a special way to interrupt.
     * @implementation Call `connection.abort()` or another interrupt method provided by the driver.
     */
    @Override
    protected void abortReadConnection(Connection connection, ResultSet resultSet) throws SQLException {
        // connection.abort(MoreExecutors.directExecutor());
    }

    /**
     * [Override as needed] Set the validation query for the Hikari connection pool. The default is SELECT 1; override if the data source has a special syntax.
     *
     * @purpose Ensure that connections obtained from the pool are valid.
     */
    @Override
    protected void setValidationQuery(HikariDataSource ds) {
        // ds.setConnectionTestQuery("SELECT 1"); // Validation query for NewDB
    }

    /**
     * [Optional override] Set system properties specific to the JDBC Driver.
     *
     * @purpose Some drivers require `System.setProperty()` to enable or disable certain features.
     */
    @Override
    protected void setJdbcDriverSystemProperties() {
        // System.setProperty("newdb.driver.property", "true");
    }
}

7.2 Регистрация NewDBJdbcExecutor в фабричном классе​

Измените doris/fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/JdbcExecutorFactory.java так, чтобы фабричный метод возвращал экземпляр NewDBJdbcExecutor в зависимости от типа источника данных.

8. Шаг 6 - Добавление регрессионных тестов​

Создайте простой тест, взяв за образец doris/regression-test/suites/external_table_p0/jdbc/type_test/select/test_mysql_all_types_select.groovy.

Создайте test_newdb_select.groovy:

suite("test_newdb_select", "p0,external,newdb") {
    String enabled = context.config.otherConfigs.get("enableJdbcTest")
    if (enabled != null && enabled.equalsIgnoreCase("true")) {
        // 1. Define the connection information for NewDB
        def newdb_port = context.config.otherConfigs.get("newdb_port")
        def driver_url = "http://your_repo/newdb-driver.jar"

        // 2. Create the Catalog
        sql """create catalog newdb_catalog properties(
            "type"="jdbc",
            "user"="root",
            "password"="123456",
            "jdbc_url" = "jdbc:newdb://\${context.config.otherConfigs.get("externalEnvIp")}:\${newdb_port}/your_db",
            "driver_url" = "\${driver_url}",
            "driver_class" = "com.newdb.jdbc.Driver"
        );"""

        // 3. Run the test
        sql """use newdb_catalog.your_db"""
        qt_select """select * from your_table order by 1 limit 10;"""

        // 4. Clean up the environment
        sql """drop catalog newdb_catalog"""
    }
}

9. Примечания по разработке​

9.1 Сопоставление типов данных​

В методе jdbcTypeToDoris уделите особое внимание следующему:

КатегорияНа что обратить внимание
Сопоставление точностиДля типа DECIMAL необходимо правильно сопоставить точность и масштаб
Типы времениУчитывайте обработку часовых поясов и точности
Особые типыНапример, некоторые особые двоичные типы

9.2 Чтение данных из отдельных источников​

В подклассах BaseJdbcExecutor сосредоточьтесь на двух методах:

МетодРоль
getColumnValueОпределяет способ получения данных из JDBC ResultSet
getOutputConverterФункция преобразования данных, выполняющая окончательное форматирование перед передачей данных на уровень C++

9.3 Обработка ошибок​

  • Преобразование исключений SQL: Преобразуйте исключения, специфичные для источника данных, в JdbcClientException.
  • Совместимость драйверов: Обрабатывайте проблемы совместимости между разными версиями драйверов.

9.4 Проталкивание предикатов и функций (необязательная оптимизация)​

Для повышения производительности запросов Doris по возможности передаёт выполнение условий WHERE и некоторых функций внешнему источнику данных:

  • Проталкивание предикатов: В большинстве случаев Doris обрабатывает это автоматически. Однако для особого синтаксиса (например, специальных функций работы с датами) может потребоваться адаптировать метод conjunctExprToString() в JdbcScanNode.java.
  • Проталкивание функций: В JdbcFunctionPushDownRule.java можно определить список функций, которые NewDB поддерживает для проталкивания, и правила замены имён функций, чтобы повысить производительность.

10. Настройка развёртывания​

10.1 Размещение драйвера​

# Place the NewDB JDBC driver in the specified directory
cp newdb-jdbc-driver.jar $DORIS_HOME/plugins/jdbc_drivers/

10.2 Создание Catalog​

CREATE CATALOG newdb_catalog PROPERTIES (
    "type" = "jdbc",
    "user" = "newdb_user",
    "password" = "newdb_password",
    "jdbc_url" = "jdbc:newdb://host:port/database",
    "driver_url" = "newdb-jdbc-driver.jar",
    "driver_class" = "com.newdb.jdbc.Driver"
);

-- Use the Catalog
USE newdb_catalog.database_name;
SELECT * FROM table_name LIMIT 10;

Частые вопросы и распространённые ошибки​

Вопрос: SHOW DATABASES не возвращает пользовательские базы данных и показывает только системные?

Ответ: Проверьте корректность списка фильтрации в getFilterInternalDatabases() и filterDatabaseNames(). Также проверьте, какой метод вызывает getDatabaseNameList(): getCatalogs() или getSchemas(). Выбор должен соответствовать семантике метаданных источника данных.

Вопрос: Из-за отсутствующих сопоставлений типов появляется Type.UNSUPPORTED?

Ответ: Добавьте соответствующие ветви в jdbcTypeToDoris. Для составных типов, таких как DECIMAL, не забудьте прочитать fieldSchema.getColumnSize() и getDecimalDigits().

Вопрос: При выполнении запросов иногда возникает OOM?

Ответ: Используйте stmt.setFetchSize() в initializeStatement, чтобы управлять количеством строк, извлекаемых за один раз. Для MySQL можно использовать Integer.MIN_VALUE, чтобы включить потоковое чтение.

Вопрос: При отмене запроса соединение не освобождается вовремя?

Ответ: Переопределите abortReadConnection и вызовите connection.abort() или другой метод прерывания, предоставляемый драйвером.

Вопрос: При создании Catalog возникает ошибка Test query failed?

Ответ: Переопределите getTestQuery(), чтобы он возвращал простейший запрос, поддерживаемый источником данных (например, для Oracle используется SELECT 1 FROM DUAL).

Report issue
На этой странице