mirror of
https://github.com/apache/doris.git
synced 2026-11-04 07:31:17 +00:00
Mingyu Chen (Rayner)
2a9245018b
### What problem does this PR solve? Related Issue: #65615 Problem Summary: Doris reads an external database through JDBC today: the source's rows become JDBC objects and then Doris columns, one value at a time, across JNI. The source is also read by a single backend, because JDBC has no notion of how a result is partitioned. This PR adds a new catalog type, `adbc`, that reads an external source through an [Arrow Database Connectivity](https://arrow.apache.org/adbc/) driver. Data arrives as Arrow record batches and is handed to BE's scanner without a per-value conversion, and a scan is split into the driver's own result partitions so that N backends read N partitions in parallel. Neither is reachable through JDBC at all. ```sql CREATE CATALOG remote_source PROPERTIES ( "type" = "adbc", "driver_url" = "libadbc_driver_flightsql.so", "uri" = "grpc://remote-doris:8070", "user" = "root", "password" = "" ); SELECT count(*) FROM remote_source.some_db.some_table; ``` Phase one targets Arrow Flight SQL sources, which makes this the intended replacement for the `remote_doris` catalog. Other sources need only their driver's `.so` and, where their SQL differs from ANSI, a dialect implementation -- no architectural change. ### Supported features What an ADBC catalog gives you from the outside -- what you can do with it and what it does for you without being asked. The boundaries are in *Known limits* further down. | Capability | What it means when you use it | Control (default) | |---|---|---| | ADBC catalog | `CREATE CATALOG ... PROPERTIES("type" = "adbc", ...)`, then query the source's tables like any other external table. Phase one targets Arrow Flight SQL sources, including another Doris cluster. | `uri`, `user`, `password` | | Arrow-native transfer | Rows arrive as Arrow batches and go straight into the scanner, with no per-value conversion on the way in -- which is where a JDBC catalog spends much of a wide or high-volume scan. | Always on | | Parallel read across backends | One scan is split into the source's own result partitions and read by several BEs at once, instead of a single BE on a single connection. Falls back to one statement when the source cannot partition. | `partitioned_read` = `auto` / `disabled` / `required` (`auto`), `max_partitions` (`1024`) | | Column pruning | Only the columns the query actually needs are requested from the source, and `EXPLAIN` prints that same statement. | Always on | | Predicate pushdown | `=`, `!=`, `<`, `<=`, `>`, `>=`, `IS [NOT] NULL`, `[NOT] IN` and `AND` / `OR` / `NOT` over them become the remote `WHERE`. Anything else -- functions, arithmetic, `LIKE`, `BETWEEN` -- stays in Doris. Doris re-applies every predicate regardless, so pushdown changes speed only, never the rows you get. | Always on | | `LIMIT` pushdown | Pushed once the whole `WHERE` was pushed, so the source never truncates ahead of a filter Doris still has to apply. | Always on | | `COUNT(*)` without column values | A count reads no column data from the source at all. | Always on | | Metadata browsing | `SHOW DATABASES`, `SHOW TABLES`, `DESC`, `SHOW CREATE TABLE` and `information_schema` over the source's own databases, tables and columns. | -- | | Automatic type mapping | Source columns arrive as Doris types, including `ARRAY` / `MAP` / `STRUCT`, `DECIMAL`, date, and timestamp with and without a zone. | -- | | Metadata cache and `REFRESH` | Name resolution and table schemas are remembered per catalog, so a query stops paying several remote round trips before it is even planned. Listings are still read live, so a table created on the source is visible with no refresh at all, and `REFRESH CATALOG` / `DATABASE` / `TABLE` drops what is remembered. | `meta.cache.adbc.metadata.enable` / `.ttl-second` (`600`) / `.capacity` (`1000`) | | The usual query surface | Joins against internal tables and against other catalogs, aggregation, `ORDER BY`, `UNION`, subqueries, `SELECT ... INTO OUTFILE`, and an MTMV built on an ADBC table. | -- | | Source SQL you can steer | The generated SQL is conservative ANSI. The connector asks the driver which vendor it is talking to, and you can override that when the answer is unhelpful. | `sql_dialect` (auto-detected, `ansi` fallback; `doris` provided) | | Driver options passed through | Anything the driver itself understands can be set on the catalog, e.g. `"adbc.adbc.snowflake.sql.db" = "..."`. | `adbc.*` properties | | Driver placement and pinning | You place the driver library (Doris ships none); a bare file name resolves under the drivers directory, and a wrong or stale build is reported at `CREATE CATALOG` rather than as a puzzling query failure later. | `driver_url`, `driver_checksum`, `driver_entrypoint` | ### Design decisions worth knowing before reviewing - **FE and BE load the same driver `.so`.** FE goes through the ADBC JNI bridge, which wraps the same C driver manager BE links statically, and BE dlopens the identical file. This is a measured constraint, not a preference: partition descriptors are driver-private bytes, and two official implementations of the same protocol already serialize incompatible messages and mis-parse each other silently rather than erroring. - **Doris ships no ADBC driver.** The library is placed by the operator, under `adbc_drivers_dir` on FE and `be/plugins/adbc_drivers` on every BE. `driver_url` therefore accepts local references only (bare name, `file://`, or an absolute path); remote schemes are rejected because a per-node download cannot promise the nodes agree. "Driver file not found" is written as a first-class error for that reason, and `driver_checksum` can pin the build. - **The SQL sent to a source is conservative ANSI by default**, generated through a dialect interface a source can claim by vendor name or by the `sql_dialect` property. Predicates are pushed all-or-nothing per conjunct from a whitelist; BE re-applies every predicate regardless, so pushdown is pure acceleration. - **Planning has side effects on a Flight SQL source**: asking the driver to partition a statement executes it. That makes this the first connector for which `EXPLAIN` had to be told not to plan for real -- see the SPI addition below. ### Changes outside the connector | Change | Why | |---|---| | `ConnectorScanRequest.isExplainOnly()` (SPI) + `PluginDrivenScanNode` fills it | `EXPLAIN` reaches `planScan` for real (its explain level is `NORMAL`, so `NereidsPlanner.distribute()` does not return early). For this connector that would execute the very query it was asked only to describe. | | `PluginDrivenScanNode` re-asks for the display statement with the current columns | The connector properties are cached in `init()`, before Nereids prunes the scan tuple, so `EXPLAIN`'s `QUERY:` line named more columns than the statement that actually runs. Affects any connector that renders remote SQL from the column list (adbc, jdbc). What goes to BE is unchanged. | | `PluginDrivenScanNode.mapFileFormatType()` learns `"arrow"` | Routes an ADBC range to BE's Arrow reader. | | `TTableFormatFileDesc.adbc_params` | The scan range's parameters; BE's ADBC reader does not read the JDBC field. | | `be/src/vec/exec/format/adbc/*` + `file_scanner_v2` gate | The BE reader. | | `FlightSqlSchemaHelper` (FE, Arrow Flight SQL server) | `GetTables` described a `DATEV2` column as date64 while BE writes date32, and described array/map/struct with placeholder children. A client that types its columns from that schema fails on the first batch. Visible to every Arrow Flight SQL client, not only this connector. | | `DataTypeTimeStampTzSerDe::read_column_from_arrow` (BE) | It never overrode the numeric serde's fixed-width path, which memcpy'd Arrow's int64 epochs into a column that stores packed date/time values -- same width, so every row was silently wrong. | | `thirdparty`: `arrow-adbc` | The C driver manager is linked into `doris_be`; the JNI bridge is built for FE (upstream's prebuilt one needs GLIBC 2.34). The SQLite and Flight SQL drivers are fetched for tests only and are **not shipped**. | | `Config.adbc_drivers_dir` / `adbc_driver_secure_path`, `conf/fe.conf` JVM options, `build.sh` | Driver placement and the FE-side plugin build/deploy wiring. | ### Behaviour to be aware of - A view on the source is not listed as a table. A Doris source ignores the base-table filter ADBC sends, so the filter is applied where the answer is read. - `partitioned_read` is `auto` by default: split the scan when the driver can, read it as one statement when it cannot. `required` forbids that downgrade, which is what keeps a test from going green while quietly exercising the fallback; `disabled` is the escape hatch. - Metadata is cached per catalog for 10 minutes by default (`meta.cache.adbc.metadata.*`), and every `REFRESH` statement drops it. A newly created remote table is reachable without any refresh. - Read only: `INSERT`, `CREATE`/`DROP TABLE` and the other write statements against an ADBC catalog are rejected. An MTMV over an ADBC table does build and refresh. ### Test coverage - `fe-connector-adbc`: 186 unit tests, no skips. About 25 of them drive the real SQLite ADBC driver through the same Java -> JNI -> C driver manager -> driver `.so` path FE takes in production; `fe-connector-api` covers the new SPI field. - BE: unit tests for the ADBC reader, the driver registry, the Arrow variant normalizer, the scanner gate and the TIMESTAMPTZ serde. - End-to-end (`regression-test/suites/external_table_p0/adbc/`): 20 suites against a live catalog -- type mapping and semantics, complex and binary types, scan edges, predicate pushdown, column pruning, query shapes, source table models, metadata operations, nested catalogs, cross-source joins, large data, outfile, MTMV, negative cases, and partitioned vs. single-statement reads compared row for row. Each suite returns early when the driver is absent, so a cluster without one does not fail. ### Known limits - Read only. No writes, DDL, statistics or aggregate pushdown. - Partitions are verified to be read completely and without duplication, but **spreading them over several backends has only been reasoned from the code**, not observed -- the test environment has one backend, and `test_adbc_multi_backend` returns early there. - `IPV4` from a Doris source arrives as a bare `INT` (both sides encode it as int32), and a source `DATETIME` arrives as `TIMESTAMPTZ`. Both are asserted, not worked around. - The FE and BE copies of the driver are not checked against each other; `driver_checksum` verifies FE's copy only. ### Release note Support ADBC catalog, which reads an external source through an Arrow Database Connectivity driver: data is transferred as Arrow record batches and a scan is split across backends by the driver's own result partitions. Phase one targets Arrow Flight SQL sources. The driver library is supplied by the operator; Doris ships none. Also fixes two defects on the Arrow Flight SQL path that are visible to existing clients: `GetTables` reported the wrong Arrow type for `DATEV2` and placeholder children for `ARRAY`/`MAP`/`STRUCT`, and a `TIMESTAMPTZ` column read from Arrow was decoded by copying its bits instead of converting the epoch.
79 lines
3.9 KiB
Plaintext
79 lines
3.9 KiB
Plaintext
# Licensed to the Apache Software Foundation (ASF) under one
|
|
# or more contributor license agreements. See the NOTICE file
|
|
# distributed with this work for additional information
|
|
# regarding copyright ownership. The ASF licenses this file
|
|
# to you under the Apache License, Version 2.0 (the
|
|
# "License"); you may not use this file except in compliance
|
|
# with the License. You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing,
|
|
# software distributed under the License is distributed on an
|
|
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
# KIND, either express or implied. See the License for the
|
|
# specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
#####################################################################
|
|
## The uppercase properties are read and exported by bin/start_fe.sh.
|
|
## To see all Frontend configurations,
|
|
## see fe/src/org/apache/doris/common/Config.java
|
|
#####################################################################
|
|
|
|
CUR_DATE=`date +%Y%m%d-%H%M%S`
|
|
|
|
# Log dir
|
|
LOG_DIR = ${DORIS_HOME}/log
|
|
|
|
# For jdk 17, this JAVA_OPTS will be used as default JVM options
|
|
JAVA_OPTS_FOR_JDK_17="-Dfile.encoding=UTF-8 -Djavax.security.auth.useSubjectCredsOnly=false -Xmx8192m -Xms8192m -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$LOG_DIR -Xlog:gc*:$LOG_DIR/fe.gc.log.$CUR_DATE:time,uptime:filecount=10,filesize=50M -Darrow.enable_null_check_for_get=false -Darrow.adbc.driver.jni.library.path=${DORIS_HOME}/lib --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/sun.nio.cs=ALL-UNNAMED --add-opens=java.base/sun.security.action=ALL-UNNAMED --add-opens=java.base/sun.util.calendar=ALL-UNNAMED --add-opens=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.management/sun.management=ALL-UNNAMED --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED --add-opens=java.xml/com.sun.org.apache.xerces.internal.jaxp=ALL-UNNAMED"
|
|
|
|
# Set your own JAVA_HOME
|
|
# JAVA_HOME=/path/to/jdk/
|
|
|
|
##
|
|
## the lowercase properties are read by main program.
|
|
##
|
|
|
|
# store metadata, must be created before start FE.
|
|
# Default value is ${DORIS_HOME}/doris-meta
|
|
# meta_dir = ${DORIS_HOME}/doris-meta
|
|
|
|
# Default dirs to put jdbc drivers,default value is ${DORIS_HOME}/jdbc_drivers
|
|
# jdbc_drivers_dir = ${DORIS_HOME}/jdbc_drivers
|
|
|
|
http_port = 8030
|
|
rpc_port = 9020
|
|
query_port = 9030
|
|
edit_log_port = 9010
|
|
arrow_flight_sql_port = 8070
|
|
|
|
# Choose one if there are more than one ip except loopback address.
|
|
# Note that there should at most one ip match this list.
|
|
# If no ip match this rule, will choose one randomly.
|
|
# use CIDR format, e.g. 10.10.10.0/24 or IP format, e.g. 10.10.10.1
|
|
# Default value is empty.
|
|
# priority_networks = 10.10.10.0/24;192.168.0.0/16
|
|
|
|
# Advanced configurations
|
|
# log_roll_size_mb = 1024
|
|
# INFO, WARN, ERROR, FATAL
|
|
sys_log_level = INFO
|
|
# NORMAL, BRIEF, ASYNC
|
|
sys_log_mode = ASYNC
|
|
# sys_log_roll_num = 10
|
|
# sys_log_verbose_modules = org.apache.doris
|
|
# audit_log_dir = $LOG_DIR
|
|
# audit_log_modules = slow_query, query
|
|
# audit_log_roll_num = 10
|
|
# meta_delay_toleration_second = 10
|
|
# qe_max_connection = 1024
|
|
# qe_query_timeout_second = 300
|
|
# qe_slow_log_ms = 5000
|
|
|
|
# Directory containing filesystem provider plugin subdirectories.
|
|
# Each subdirectory is one storage backend (e.g., s3/, hdfs/, azure/).
|
|
# If empty, only classpath-based built-in providers are used.
|
|
# filesystem_plugin_root = ${DORIS_HOME}/plugins/filesystem
|