diff --git a/community/developer-guide/data-lineage-plugin-development.md b/community/developer-guide/data-lineage-plugin-development.md new file mode 100644 index 0000000000000..a7bb16569e79f --- /dev/null +++ b/community/developer-guide/data-lineage-plugin-development.md @@ -0,0 +1,744 @@ +--- +title: Data Lineage Plugin Development +language: en +description: Build, package, deploy, and verify an Apache Doris data lineage plugin that logs lineage events or forwards them to an external governance system. +keywords: + - Apache Doris + - data lineage + - LineagePlugin + - LineagePluginFactory + - ServiceLoader + - plugin development +--- + + + + + + +This document describes how to develop an external data lineage plugin for Apache Doris. The plugin receives a `LineageInfo` event after a supported DML statement succeeds, then converts the event and writes it to logs, a messaging system, or an external metadata governance platform. + +> Applicable version: Apache Doris community edition 4.0.6 and later. Compile the plugin against the exact Doris community release or source revision deployed on the FE. + +## Scope and lifecycle + +The lineage framework is an FE-side SPI. It does not store lineage data, provide a query API, or render a lineage graph. A plugin converts the in-memory Java objects in `LineageInfo` to the format and transport protocol required by its downstream system. + +The current implementation generates events only for successful `INSERT INTO ... SELECT`, `INSERT OVERWRITE TABLE ... SELECT`, and `CREATE TABLE AS SELECT` statements. `VALUES`-only writes and writes whose target is `__internal_schema` are skipped. `SELECT`, `UPDATE`, `DELETE`, and load jobs are outside the scope of this framework. Some `UPDATE` and `DELETE` execution paths internally reuse an Insert command, but the original command-type check excludes them from lineage event submission. + +![Data lineage collection architecture: successful supported DML statements are analyzed by Nereids, extracted into LineageInfo, queued in FE, and delivered by a plugin to an external governance system.](/images/data-lineage/lineage-architecture.svg) + +`eventFilter()` is called once before lineage extraction on the DML query path and again before worker dispatch. Return `false` when the plugin should not receive events. The method is called concurrently from query threads and the worker thread, so it must be thread-safe. `exec()` is called by one worker thread, but can run concurrently with `eventFilter()`. + +At FE startup, ServiceLoader first discovers each `LineagePluginFactory`. FE then calls the Factory's `create(context)` method and calls `initialize(context)` on the returned plugin. The default `LineagePluginFactory.create(context)` implementation delegates to the no-argument `create()`, so the minimal example in this guide implements only `create()`. `PluginContext` exposes an immutable property map through `getProperties()`; FE stores the plugin name and directory in the `plugin.name` and `plugin.path` entries. Access them with `context.getProperties().get("plugin.name")` and `context.getProperties().get("plugin.path")`. A production plugin that reads a configuration file from its plugin directory can override `initialize(context)`, locate the file through the `plugin.path` entry, and initialize its resources there. + +## Lineage event model + +`LineageInfo` contains a target table, target output columns, source tables, direct lineage, indirect lineage, and a `LineageContext`. + +| Part | Meaning | +| --- | --- | +| Target | The target table and output columns of the write. | +| Source tables | Tables referenced by the analyzed logical plan. | +| Direct lineage | A map from each output Slot to source expressions, classified as `IDENTITY`, `TRANSFORMATION`, or `AGGREGATION`. | +| Dataset indirect lineage | Expressions that affect the complete result set: `JOIN`, `FILTER`, `GROUP_BY`, and `SORT`. | +| Output indirect lineage | Dependencies that affect a specific output column: `WINDOW` and `CONDITIONAL`. | +| Query context | Query ID, SQL text, user, client IP, session database and catalog, execution state, timestamp, duration, and sanitized external catalog properties. | + +Direct lineage describes how an output value is produced. A plain source-column reference is `IDENTITY`, for example `target.customer_id <- source.customer_id`. A calculation, function, window expression, or conditional expression without an aggregate function is `TRANSFORMATION`, for example `UPPER(source.region)`. An expression that contains an aggregate function is `AGGREGATION`, for example `SUM(source.amount)`. Indirect lineage records expressions that affect the output without directly becoming an output value, including Join keys, `WHERE` and `HAVING` predicates, grouping keys, and sorting keys. `WINDOW` records window partitioning and ordering inputs. `CONDITIONAL` records the effect of `CASE`, `IF`, or `COALESCE` on a specific output column. + +### Lineage event model example + +The following example uses one source table and one target table to cover all three direct-lineage types, dataset indirect lineage, and output indirect lineage. The current user must have privileges to create a database and tables, write data, and query data. + +```sql +CREATE DATABASE IF NOT EXISTS lineage_demo; +USE lineage_demo; + +CREATE TABLE lineage_source ( + region VARCHAR(32), + amount DECIMAL(18, 2), + status VARCHAR(16) +) +DUPLICATE KEY(region) +DISTRIBUTED BY HASH(region) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_target ( + region VARCHAR(32), + total_amount DECIMAL(18, 2), + region_seq BIGINT, + region_group VARCHAR(16) +) +DUPLICATE KEY(region) +DISTRIBUTED BY HASH(region) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +INSERT INTO lineage_source VALUES + ('east', 100.00, 'PAID'), + ('east', 30.00, 'CANCELLED'), + ('west', 80.00, 'PAID'); + +INSERT INTO lineage_target +SELECT + region, + SUM(amount) AS total_amount, + ROW_NUMBER() OVER (ORDER BY region) AS region_seq, + CASE + WHEN region = 'east' THEN 'CORE' + ELSE 'OTHER' + END AS region_group +FROM lineage_source +WHERE status = 'PAID' +GROUP BY region +ORDER BY SUM(amount) DESC; + +SELECT region, total_amount, region_seq, region_group +FROM lineage_target +ORDER BY region; +``` + +The query returns: + +```text ++--------+--------------+------------+--------------+ +| region | total_amount | region_seq | region_group | ++--------+--------------+------------+--------------+ +| east | 100.00 | 1 | CORE | +| west | 80.00 | 2 | OTHER | ++--------+--------------+------------+--------------+ +``` + +The first, `VALUES`-only Insert does not generate an event. After `INSERT INTO lineage_target ... SELECT` succeeds, its event contains these key relationships: + +| Lineage level | Content in this example | +| --- | --- | +| Table lineage | `internal.lineage_demo.lineage_source` -> `internal.lineage_demo.lineage_target`. | +| Direct column lineage | `region` is `IDENTITY`; `total_amount` is `AGGREGATION`; `region_seq` and `region_group` are `TRANSFORMATION`. | +| Dataset indirect lineage | `FILTER` is `status = 'PAID'`; `GROUP_BY` is `region`; `SORT` is `SUM(amount) DESC`. | +| Output indirect lineage | The `WINDOW` dependency of `region_seq` is `region`; the `CONDITIONAL` dependency of `region_group` is `region = 'east'`. | +| Query context | Contains the actual Query ID, the Insert SQL above, execution user, and client IP. The session database is `lineage_demo`, the Internal Catalog is `internal`, and the execution state is `OK`. | + +This example has no Join and therefore no dataset-level `JOIN` dependency. A query with a Join records the Join condition as a dataset-level `JOIN` dependency. + +Before the event reaches the plugin, the extractor resolves CTE producer expressions and expands `UNION` branches. A plugin should still treat `Expression`, `SlotReference`, and `TableIf` as Doris internal Java objects and convert them to stable names or a downstream schema before sending them outside FE. + +## Plugin API reference + +The lineage SPI consists of a Factory discovered at FE startup, a plugin instance created by that Factory, an immutable runtime context, and the `LineageInfo` event passed to the plugin. The following summary lists the extension methods that plugin implementations use directly. + +### API summary + +| Method | Input | Return type | Purpose | +| --- | --- | --- | --- | +| `LineagePluginFactory.name()` | None | `String` | Return the unique Factory name used by `activate_lineage_plugin`. | +| `LineagePluginFactory.create()` | None | `LineagePlugin` | Create a plugin instance without reading the runtime context in the Factory. | +| `LineagePluginFactory.create(PluginContext)` | `PluginContext` | `LineagePlugin` | Create a plugin instance with access to the runtime context. The default implementation delegates to `create()`. | +| `LineagePlugin.name()` | None | `String` | Return the plugin name. It should match the Factory name. | +| `LineagePlugin.eventFilter()` | None | `boolean` | Indicate whether the plugin currently wants lineage events. | +| `LineagePlugin.exec(LineageInfo)` | `LineageInfo` | `boolean` | Process one lineage event. | +| `Plugin.initialize(PluginContext)` | `PluginContext` | `void` | Initialize resources after the Factory creates the plugin. The default implementation does nothing. | +| `Plugin.close()` | None | `void` | Lifecycle hook intended to release plugin resources. The default implementation does nothing. | +| `PluginContext.getProperties()` | None | `Map` | Return the immutable runtime property map. | + +### `LineagePluginFactory` + +`LineagePluginFactory` is the ServiceLoader entry point. FE discovers the Factory, filters it by the value returned from `name()`, and then creates one plugin instance. + +```java +String name(); +LineagePlugin create(); +default LineagePlugin create(PluginContext context); +``` + +#### `name()` + +- **Input:** none. +- **Returns:** `String`, the unique, case-sensitive Factory identifier. +- **Behavior:** the value is matched against each item in `activate_lineage_plugin`. The example returns `example-lineage`. + +#### `create()` + +- **Input:** none. +- **Returns:** a new `LineagePlugin` instance. +- **Behavior:** implement this method when plugin construction does not require the runtime context. The default `create(PluginContext)` method delegates to it. + +#### `create(PluginContext context)` + +- **Input:** `PluginContext context`, the immutable runtime context prepared for this plugin directory. +- **Returns:** a new `LineagePlugin` instance. +- **Behavior:** override this method only when the Factory itself needs runtime properties during construction. FE calls `initialize(context)` on the returned plugin after creation. + +Returning `null` or throwing during creation prevents that plugin from loading. Do not perform long-running downstream calls in a Factory method because plugin discovery happens during FE startup. + +### `LineagePlugin` + +`LineagePlugin` receives lineage events and inherits the `initialize()` and `close()` lifecycle methods from `Plugin`. + +```java +String name(); +boolean eventFilter(); +boolean exec(LineageInfo lineageInfo); +default void initialize(PluginContext context); +default void close(); +``` + +#### `name()` + +- **Input:** none. +- **Returns:** `String`, the plugin identifier. +- **Behavior:** return the same stable identifier as `LineagePluginFactory.name()` so activation, logs, and operational configuration use one name. + +#### `eventFilter()` + +- **Input:** none. +- **Returns:** `boolean`; `true` means the plugin wants lineage events, and `false` means FE skips it. +- **Behavior:** FE calls this method before lineage extraction on query threads and again before dispatch on the lineage worker thread. It must be thread-safe and should return quickly. If every loaded plugin returns `false` before extraction, FE skips extraction for that statement. + +#### `exec(LineageInfo lineageInfo)` + +- **Input:** `LineageInfo lineageInfo`, one extracted lineage event from a successful supported DML statement. +- **Returns:** `boolean`; `true` reports success and `false` reports failure to the caller. +- **Behavior:** one lineage worker thread invokes `exec()`, but it can run concurrently with `eventFilter()` calls on query threads. The current framework does not retry, requeue, or change the DML state when `exec()` returns `false`. An exception is logged and processing continues with later plugins or events. + +#### `initialize(PluginContext context)` + +- **Input:** `PluginContext context`, the same runtime context supplied to the Factory. +- **Returns:** `void`. +- **Behavior:** FE calls this method after creating the plugin. Override it to read configuration from `plugin.path`, create clients, or allocate other resources. Throwing prevents the plugin from loading. + +#### `close()` + +- **Input:** none. +- **Returns:** `void`. +- **Behavior:** this inherited lifecycle hook is intended for resource cleanup. The current lineage processor has no dynamic unload or reload path, so a plugin must not rely on `close()` being called when a JAR is replaced. Replacing a JAR still requires an FE restart. + +### `PluginContext` + +```java +Map getProperties(); +``` + +`getProperties()` takes no input and returns an immutable `Map`. The lineage loader provides these entries: + +| Property | Type | Meaning | +| --- | --- | --- | +| `plugin.name` | `String` | The name returned by the discovered Factory. | +| `plugin.path` | `String` | The absolute path of the external plugin directory. | + +Read them with `context.getProperties().get("plugin.name")` and `context.getProperties().get("plugin.path")`. The map cannot be modified. A plugin can use `plugin.path` to locate its own configuration files, but must not assume that the directory name equals the Factory name. + +### `LineageInfo` + +`LineageInfo` is the input type of `exec()`. Plugins should read it through the following methods and convert Doris objects to their downstream event schema. + +| Method | Return type | Meaning | +| --- | --- | --- | +| `getTargetTable()` | `TableIf` | Target table of the write. | +| `getTargetColumns()` | `List` | Target output columns in write order. | +| `getTableLineageSet()` | `Set` | Source tables referenced by the analyzed plan. | +| `getDirectLineageMap()` | `Map>` | Direct source expressions for each output Slot. | +| `getDatasetIndirectLineageMap()` | `Multimap` | Dataset-level `JOIN`, `FILTER`, `GROUP_BY`, and `SORT` dependencies. | +| `getInDirectLineageMapByDataset()` | `Map>` | A derived view that applies the dataset-level dependencies to every output Slot. It excludes `WINDOW` and `CONDITIONAL`. | +| `getOutputIndirectLineageMap()` | `Map>` | Per-output `WINDOW` and `CONDITIONAL` dependencies. | +| `getContext()` | `LineageContext` | Query and execution metadata for this event. | + +`LineageInfo` also exposes setters and `add*` methods used by the FE extractor. A sink plugin normally consumes the object through the getters above and should not mutate the event. The returned `TableIf`, `Slot`, `SlotReference`, and `Expression` values are Doris runtime objects, not stable wire-format identifiers. + +### `LineageContext` + +`LineageContext` is returned by `LineageInfo.getContext()`. Its getters take no input. + +| Method | Return type | Meaning | +| --- | --- | --- | +| `getSourceCommand()` | `Class` | Command class that produced the event, such as `InsertIntoTableCommand`. | +| `getQueryId()` | `String` | Query ID. | +| `getQueryText()` | `String` | Original DML text. | +| `getUser()` | `String` | Executing user. | +| `getClientIp()` | `String` | Client IP address. | +| `getState()` | `String` | Query execution state. | +| `getDatabase()` | `String` | Session database; it is not necessarily the target-table database. | +| `getCatalog()` | `String` | Session Catalog; it is not necessarily the target-table Catalog. | +| `getTimestampMs()` | `long` | Event timestamp in milliseconds, or `-1` when unavailable. | +| `getDurationMs()` | `long` | Query duration in milliseconds, or `-1` when unavailable. | +| `getExternalCatalogProperties()` | `Map>` | Sanitized properties for external Catalogs referenced by the query. | + +The original SQL, user, client IP, expressions, and external Catalog properties can contain sensitive data. A production plugin should apply its own access control, redaction, size limits, and retention policy before exporting them. + +## Develop a plugin + +This section uses the public `4.0.6` tag from the official `github.com/apache/doris` community repository and adds an `example-lineage` Maven submodule inside the source tree. The plugin writes each event only to the FE log. It does not start an HTTP service or depend on an external governance platform. This implementation demonstrates the complete plugin development and loading flow. A production plugin can convert `LineageInfo` to a third-party schema, call an API exposed by a metadata platform such as OpenMetadata, or emit lineage events that conform to the OpenLineage specification. All source paths below are relative to the same Doris source root. + +### Step 1: Prepare the Doris source + +Check out the public release that exactly matches the target FE, and record the source root in `DORIS_SOURCE`: + +```shell +git clone https://github.com/apache/doris.git +export DORIS_SOURCE="$(pwd)/doris" +cd "${DORIS_SOURCE}" +git checkout 4.0.6 +``` + +Every subsequent `$DORIS_SOURCE/fe_plugins/...` path refers to this source directory. Do not replace it with the FE installation directory, and do not compile against FE JARs from another version. + +### Step 2: Create the Maven modules + +Run the following commands from the Doris source root: + +```shell +cd "${DORIS_SOURCE}" +mkdir -p fe_plugins/lineage/example-lineage/src/main/java/org/apache/doris/plugin/lineage/example +mkdir -p fe_plugins/lineage/example-lineage/src/main/resources/META-INF/services +``` + +After this section, the new files have the following layout: + +```text +$DORIS_SOURCE/ +└── fe_plugins/ + ├── pom.xml # Existing file; add the lineage module + └── lineage/ + ├── pom.xml + └── example-lineage/ + ├── pom.xml + └── src/main/ + ├── java/org/apache/doris/plugin/lineage/example/ + │ ├── ExampleLineagePluginFactory.java + │ └── LoggingLineagePlugin.java + └── resources/META-INF/services/ + └── org.apache.doris.nereids.lineage.LineagePluginFactory +``` + +Edit the existing `$DORIS_SOURCE/fe_plugins/pom.xml` and add `lineage` to its ``. Keep all existing modules; do not replace the complete file with this fragment: + +```xml + + + lineage + +``` + +Create `$DORIS_SOURCE/fe_plugins/lineage/pom.xml`: + +```xml + + + + 4.0.0 + + org.apache.doris + fe-plugins + 1.0-SNAPSHOT + ../pom.xml + + lineage + pom + + example-lineage + + +``` + +Create `$DORIS_SOURCE/fe_plugins/lineage/example-lineage/pom.xml`: + +```xml + + + + 4.0.0 + + org.apache.doris + lineage + 1.0-SNAPSHOT + ../pom.xml + + example-lineage + jar + + + org.apache.doris + fe-core + ${doris.version} + provided + + + org.apache.doris + fe-extension-spi + ${doris.version} + provided + + + org.apache.logging.log4j + log4j-api + provided + + + + example-lineage + + +``` + +All dependencies use `provided` because FE supplies these APIs. Do not package Doris or Log4j classes in the plugin JAR. + +### Step 3: Implement the Factory + +Create `$DORIS_SOURCE/fe_plugins/lineage/example-lineage/src/main/java/org/apache/doris/plugin/lineage/example/ExampleLineagePluginFactory.java`: + +```java +// 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. + +package org.apache.doris.plugin.lineage.example; + +import org.apache.doris.nereids.lineage.LineagePlugin; +import org.apache.doris.nereids.lineage.LineagePluginFactory; + +/** Factory discovered by ServiceLoader at FE startup. */ +public class ExampleLineagePluginFactory implements LineagePluginFactory { + @Override + public String name() { + return "example-lineage"; + } + + @Override + public LineagePlugin create() { + return new LoggingLineagePlugin(); + } +} +``` + +The `example-lineage` value returned by `name()` is the unique plugin identifier. The later `activate_lineage_plugin` setting must use exactly the same value. + +### Step 4: Implement the logging plugin + +Create `$DORIS_SOURCE/fe_plugins/lineage/example-lineage/src/main/java/org/apache/doris/plugin/lineage/example/LoggingLineagePlugin.java`: + +```java +// 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. + +package org.apache.doris.plugin.lineage.example; + +import org.apache.doris.catalog.TableIf; +import org.apache.doris.nereids.lineage.LineageContext; +import org.apache.doris.nereids.lineage.LineageInfo; +import org.apache.doris.nereids.lineage.LineagePlugin; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.Objects; +import java.util.stream.Collectors; + +/** Minimal lineage plugin that writes one summary line for each event. */ +public class LoggingLineagePlugin implements LineagePlugin { + private static final Logger LOG = LogManager.getLogger(LoggingLineagePlugin.class); + private static final String PLUGIN_NAME = "example-lineage"; + + @Override + public String name() { + return PLUGIN_NAME; + } + + @Override + public boolean eventFilter() { + return true; + } + + @Override + public boolean exec(LineageInfo lineageInfo) { + if (lineageInfo == null) { + return false; + } + + LineageContext context = lineageInfo.getContext(); + LOG.info("LINEAGE_EVENT queryId={} command={} user={} clientIp={} " + + "catalog={} database={} state={} durationMs={} target={} sources={} " + + "direct={} datasetIndirect={} outputIndirect={}", + value(context == null ? null : context.getQueryId()), + command(context), + value(context == null ? null : context.getUser()), + value(context == null ? null : context.getClientIp()), + value(context == null ? null : context.getCatalog()), + value(context == null ? null : context.getDatabase()), + value(context == null ? null : context.getState()), + context == null ? -1L : context.getDurationMs(), + tableName(lineageInfo.getTargetTable()), + sourceTables(lineageInfo), + lineageInfo.getDirectLineageMap(), + lineageInfo.getDatasetIndirectLineageMap(), + lineageInfo.getOutputIndirectLineageMap()); + return true; + } + + private String sourceTables(LineageInfo lineageInfo) { + if (lineageInfo.getTableLineageSet() == null) { + return "[]"; + } + return lineageInfo.getTableLineageSet().stream() + .filter(Objects::nonNull) + .map(TableIf::getNameWithFullQualifiers) + .sorted() + .collect(Collectors.joining(",", "[", "]")); + } + + private String tableName(TableIf table) { + return table == null ? "" : table.getNameWithFullQualifiers(); + } + + private String command(LineageContext context) { + return context == null || context.getSourceCommand() == null + ? "" : context.getSourceCommand().getSimpleName(); + } + + private String value(String value) { + return value == null ? "" : value; + } +} +``` + +This example does not log the original SQL, external catalog properties, or authentication information. Direct and indirect lineage expressions can still contain SQL literals, so use this implementation only for development validation. A production plugin should sanitize expressions according to its security policy, limit the length of each log entry, and decide whether to record user and client IP values. + +### Step 5: Register the Factory + +Create `$DORIS_SOURCE/fe_plugins/lineage/example-lineage/src/main/resources/META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory`. The file contains only the fully qualified Factory class name: + +```text +org.apache.doris.plugin.lineage.example.ExampleLineagePluginFactory +``` + +Each external plugin directory can expose only one `LineagePluginFactory`. Factory names must also be unique across all loaded plugins. + +### Step 6: Build the plugin + +Before the first plugin build in this source directory, prepare the Doris source build environment for [Linux](/community/source-install/compilation-linux) or [macOS](/community/source-install/compilation-mac), then build FE once. The following environment variables disable only the UI, Hive UDF, and BE Java extensions that the lineage plugin build does not require. They do not skip FE: + +```shell +cd "${DORIS_SOURCE}" +DISABLE_BUILD_UI=ON \ +DISABLE_BE_JAVA_EXTENSIONS=ON \ +DISABLE_BUILD_HIVE_UDF=ON \ +./build.sh --fe +``` + +This step generates the Thrift and Protobuf sources and builds FE. Do not skip it and run the Maven command below directly in a newly checked-out source tree; compilation fails because generated classes such as `org.apache.doris.thrift` are missing. You do not need to repeat it after `./build.sh --fe` has already succeeded for the same source version. + +Next, install the FE modules required by the plugin into the local Maven repository, then build the plugin submodule: + +```shell +cd "${DORIS_SOURCE}" +mvn -f fe/pom.xml install \ + -pl fe-common,fe-extension-spi,fe-extension-loader,fe-core \ + -DskipTests -Dcheckstyle.skip=true +mvn -f fe_plugins/pom.xml -pl lineage/example-lineage -am clean package \ + -DskipTests -Dcheckstyle.skip=true +``` + +A successful build creates: + +```text +$DORIS_SOURCE/fe_plugins/lineage/example-lineage/target/example-lineage.jar +``` + +Inspect the SPI file and JAR contents: + +```shell +PLUGIN_JAR="${DORIS_SOURCE}/fe_plugins/lineage/example-lineage/target/example-lineage.jar" +unzip -p "${PLUGIN_JAR}" \ + META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory +if jar tf "${PLUGIN_JAR}" | grep -qE '^org/apache/doris/(nereids|extension)/'; then + echo "ERROR: Doris API classes must not be packaged in the plugin JAR" >&2 + exit 1 +fi +``` + +The first command must print `org.apache.doris.plugin.lineage.example.ExampleLineagePluginFactory`. The subsequent check prints nothing and returns zero when the JAR does not contain Doris API classes. + +### Step 7: Deploy and activate the plugin + +In this section, `FE_HOME` denotes one FE installation directory containing `bin/`, `conf/`, and `lib/`, not the Doris source directory. `start_fe.sh` sets this directory as the FE process's `DORIS_HOME`. Copy the JAR to every FE: + +```shell +export FE_HOME=/opt/apache-doris/fe +mkdir -p "${FE_HOME}/plugins/lineage/example-lineage" +cp "${DORIS_SOURCE}/fe_plugins/lineage/example-lineage/target/example-lineage.jar" \ + "${FE_HOME}/plugins/lineage/example-lineage/" +``` + +The operating-system user that runs FE must be able to read the plugin directory and JAR. + +The deployed layout is: + +```text +$FE_HOME/ +└── plugins/ + └── lineage/ + └── example-lineage/ + └── example-lineage.jar +``` + +Edit `$FE_HOME/conf/fe.conf` on every FE: + +```text +plugin_dir = /opt/apache-doris/fe/plugins +activate_lineage_plugin = example-lineage +lineage_event_queue_size = 50000 +``` + +| Configuration | Type and default | Required | Effect | +| --- | --- | --- | --- | +| `plugin_dir` | String; `$FE_HOME/plugins` | No | Plugin root directory. The lineage loader scans its direct children under `lineage/`. Omit this setting when using the default directory. | +| `activate_lineage_plugin` | String array; empty | Explicit configuration recommended | Comma-separated Factory names to instantiate. An empty value instantiates every discovered Factory. | +| `lineage_event_queue_size` | Positive integer; `50000` | No | Maximum events waiting for the worker on the current FE. New events are discarded when the queue is full. | + +These settings apply to the current FE process and are read only at startup. Deploy and configure the plugin on every FE that can execute DML. Restart the corresponding FE after changing a setting or replacing the JAR. The settings cannot be changed dynamically through `ADMIN SET FRONTEND CONFIG`. + +## Verify the plugin + +After restarting FE, first confirm that the plugin loaded. If the FE log directory was customized, replace `FE_LOG` with the actual `fe.log` path: + +```shell +export FE_LOG="${FE_HOME}/log/fe.log" +grep 'Loaded lineage plugin: example-lineage' "${FE_LOG}" | tail -1 +``` + +Run the preceding [lineage event model example](#lineage-event-model-example), then query the latest event: + +```shell +grep 'LINEAGE_EVENT' "${FE_LOG}" | tail -1 +``` + +The log must contain the actual Query ID, `InsertIntoTableCommand`, target table `lineage_target`, source table `lineage_source`, and the `direct`, `datasetIndirect`, and `outputIndirect` sections. For example: + +```text +LINEAGE_EVENT queryId= command=InsertIntoTableCommand ... +target=internal.lineage_demo.lineage_target +sources=[internal.lineage_demo.lineage_source] direct={...} +datasetIndirect={...} outputIndirect={...} +``` + +The actual log is one line; it is wrapped above for readability. `INSERT INTO lineage_source VALUES ...` does not generate an event. Only `INSERT INTO lineage_target ... SELECT` produces this log entry. + +When validation is complete and the example data is no longer needed, run: + +```sql +DROP DATABASE lineage_demo; +``` + +## Reliability and troubleshooting + +| Condition | Framework behavior | Plugin recommendation | +| --- | --- | --- | +| Queue is full | The new event is discarded and a warning is logged. The DML result is unaffected. | Keep `exec()` fast; batch or buffer downstream writes outside the FE critical path. | +| `exec()` throws | The worker logs the exception and continues with the next plugin or event. | Catch expected downstream failures, add bounded retries, and expose metrics. | +| `exec()` returns `false` | The current framework ignores the return value. It does not retry, requeue the event, or change the DML state. | Record the failure and implement bounded retries inside the plugin. Do not rely on the return value for recovery. | +| Plugin is slow | One worker dispatches events serially. A slow plugin delays every following event. | Use a bounded internal queue and a separate sender pool for a slow downstream system. | +| Duplicate Factory name | The first discovered Factory is retained and the duplicate is skipped. | Use a globally unique name. | +| JAR changes after startup | The loader does not watch, reload, or unload plugin directories. | Restart FE after replacing JARs or dependencies. | + +### Diagnose problems from FE logs + +| Log keyword | Common cause | Action | +| --- | --- | --- | +| `Skip lineage plugin directory due to load failure` | The plugin directory has no JAR, the SPI file is missing, the directory exposes multiple factories, or dependency loading failed. | Inspect the directory and JAR, confirm that the SPI file contains one Factory class name, and inspect `stage` and `message` in the exception. | +| `Skip lineage plugin not in activate_lineage_plugin` | The Factory was discovered, but its `name()` is not in the activation list. | Compare the Factory name with `activate_lineage_plugin`. Names are case-sensitive. | +| `Failed to create/initialize lineage plugin` | `create(context)` or `initialize(context)` threw an exception. | Inspect the stack trace, initialization logic, configuration files, and directory permissions. | +| `the lineage event queue is full` | The plugin consumes events more slowly than they are produced, and the current FE queue is full. | First shorten `exec()` processing time and investigate downstream blocking, then evaluate whether to increase `lineage_event_queue_size`. | + +### Disable and roll back the plugin + +To disable the plugin, stop the corresponding FE, move the plugin directory out of `plugin_dir/lineage/`, remove `example-lineage` from `activate_lineage_plugin`, and start FE. Do not only clear `activate_lineage_plugin` while leaving the JAR in place, because an empty list activates every discovered lineage plugin. + +```shell +mkdir -p "${FE_HOME}/plugins-disabled" +mv "${FE_HOME}/plugins/lineage/example-lineage" \ + "${FE_HOME}/plugins-disabled/" +``` + +After FE starts, confirm that no new `Loaded lineage plugin: example-lineage` entry appears. To restore the plugin, move the directory back under `${FE_HOME}/plugins/lineage/`, restore `activate_lineage_plugin`, and restart FE again. + +Lineage delivery is best effort and is not transactional with the DML. A successful DML does not guarantee delivery to an external governance system. Use the Query ID and target table to construct idempotent downstream event identifiers. diff --git a/docs/data-governance/data-lineage.md b/docs/data-governance/data-lineage.md new file mode 100644 index 0000000000000..5dc76226b9955 --- /dev/null +++ b/docs/data-governance/data-lineage.md @@ -0,0 +1,336 @@ +--- +{ + "title": "Data Lineage Technical Guide", + "sidebar_label": "Data Lineage", + "language": "en", + "description": "Collect table-level and column-level lineage from supported Doris DML statements through an external LineagePlugin.", + "keywords": ["Apache Doris", "data lineage", "data governance", "LineagePlugin", "column-level lineage"] +} +--- + + + + +Data Lineage extracts table-level and column-level dependencies from supported Doris DML statements and sends them to external governance systems through a `LineagePlugin`. Doris is the lineage producer. It does not retain events, expose a lineage query API, or provide lineage visualization. + +> This capability is available in the Dev version. It was first released in Apache Doris 4.0.6. + +## Capabilities and limitations + +The framework generates an event only after one of the following statements succeeds: + +| Supported statement | Event behavior | +| --- | --- | +| `INSERT INTO ... SELECT` | Generates one event after the Insert succeeds. | +| `INSERT OVERWRITE TABLE ... SELECT` | Generates one event after the Overwrite succeeds. | +| `CREATE TABLE AS SELECT` | Generates one event after the internal Insert succeeds. | + +The current implementation does not generate events for `SELECT`, `UPDATE`, `DELETE`, load jobs, `VALUES`-only inserts, or writes whose target is `__internal_schema`. Some `UPDATE` and `DELETE` execution paths internally reuse an Insert command, but the original command-type check excludes them from lineage event submission. + +:::caution Delivery guarantee + +Lineage delivery is asynchronous and best effort. A successful DML statement is not rolled back when event collection or plugin delivery fails. Events can be discarded when the queue is full, and Doris does not retry or persist them. + +::: + +## Technical design + +### Collection flow + +For a supported statement, Doris records the analyzed Nereids logical plan in an `afterAnalyze` planner hook. After the DML succeeds, Doris extracts lineage from that plan and submits the event to an FE-local queue. A single daemon worker evaluates every loaded plugin and dispatches the event to each plugin whose `eventFilter()` returns `true`. + +![Data lineage collection architecture: successful supported DML statements are analyzed by Nereids, extracted into LineageInfo, queued in FE, and delivered by a plugin to an external governance system.](/images/data-lineage/lineage-architecture.svg) + +Before extraction, Doris calls `eventFilter()` on the loaded plugins. If no plugin is willing to receive events, extraction is skipped. The worker evaluates `eventFilter()` again before dispatch. + +### Lineage content + +Each `LineageInfo` event contains table lineage, direct column lineage, two kinds of indirect lineage, and query context. The complete SQL example below establishes the source tables, target table, and result used by the following sections. + +#### Example SQL and result + +The current user must have privileges to create a database and tables, write data, and query data, and the FE must have loaded a lineage plugin. The example uses two source tables and creates a customer summary through a CTE, Join, filters, aggregation, a window function, a conditional expression, and sorting. + +```sql +CREATE DATABASE IF NOT EXISTS lineage_demo; +USE lineage_demo; + +CREATE TABLE lineage_orders ( + order_id BIGINT, + customer_id BIGINT, + region VARCHAR(16), + amount DECIMAL(18, 2), + status VARCHAR(16) +) +DUPLICATE KEY(order_id) +DISTRIBUTED BY HASH(order_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_customers ( + customer_id BIGINT, + customer_name VARCHAR(64), + customer_level VARCHAR(16) +) +DUPLICATE KEY(customer_id) +DISTRIBUTED BY HASH(customer_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_customer_summary ( + customer_id BIGINT, + region_label VARCHAR(16), + total_amount DECIMAL(18, 2), + customer_seq BIGINT, + region_group VARCHAR(16) +) +DUPLICATE KEY(customer_id) +DISTRIBUTED BY HASH(customer_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +INSERT INTO lineage_customers VALUES + (1, 'Alice', 'VIP'), + (2, 'Bob', 'VIP'), + (3, 'Carol', 'REGULAR'); + +INSERT INTO lineage_orders VALUES + (101, 1, 'east', 100.00, 'PAID'), + (102, 1, 'east', 50.00, 'PAID'), + (103, 2, 'west', 80.00, 'PAID'), + (104, 2, 'west', 30.00, 'CANCELLED'), + (105, 3, 'north', 200.00, 'PAID'); + +INSERT INTO lineage_customer_summary +WITH customer_totals AS ( + SELECT + o.customer_id, + UPPER(o.region) AS region_label, + SUM(o.amount) AS total_amount + FROM lineage_orders o + JOIN lineage_customers c + ON o.customer_id = c.customer_id + WHERE o.status = 'PAID' + AND c.customer_level = 'VIP' + GROUP BY o.customer_id, UPPER(o.region) + HAVING SUM(o.amount) >= 50 +) +SELECT + customer_id, + region_label, + total_amount, + ROW_NUMBER() OVER (ORDER BY customer_id) AS customer_seq, + CASE + WHEN region_label = 'EAST' THEN 'CORE' + ELSE 'OTHER' + END AS region_group +FROM customer_totals +ORDER BY total_amount DESC; + +SELECT customer_id, region_label, total_amount, customer_seq, region_group +FROM lineage_customer_summary +ORDER BY customer_id; +``` + +The query returns: + +```text ++-------------+--------------+--------------+--------------+--------------+ +| customer_id | region_label | total_amount | customer_seq | region_group | ++-------------+--------------+--------------+--------------+--------------+ +| 1 | EAST | 150.00 | 1 | CORE | +| 2 | WEST | 80.00 | 2 | OTHER | ++-------------+--------------+--------------+--------------+--------------+ +``` + +The two `VALUES`-only inserts do not generate lineage events. The `INSERT INTO lineage_customer_summary ... SELECT` statement generates one `LineageInfo` event, whose contents are described below. + +#### Table lineage + +Table lineage records one target table and every source table scanned by the analyzed logical plan. The query that writes `lineage_customer_summary` scans both `lineage_orders` and `lineage_customers`, so both source tables have a table-level relationship with the target. A CTE is resolved to its underlying tables, and every `UNION` branch contributes to the source-table set. + +This example produces the following table lineage: + +| Source table | Target table | +| --- | --- | +| `lineage_orders` | `lineage_customer_summary` | +| `lineage_customers` | `lineage_customer_summary` | + +#### Direct column lineage + +Direct column lineage describes which source expression produces each target-column value. After resolving aliases and CTEs, Doris classifies expressions in this order: an expression containing an aggregate function is `AGGREGATION`; a plain source-column reference is `IDENTITY`; every other expression is `TRANSFORMATION`. + +| Type | Meaning | Example | +| --- | --- | --- | +| `IDENTITY` | The target value is taken directly from a source column without a function or operation. | `lineage_customer_summary.customer_id` comes directly from `lineage_orders.customer_id`. | +| `TRANSFORMATION` | A non-aggregate expression transforms the target value, such as arithmetic, a string function, a window function, or a conditional expression. | `region_label` comes from `UPPER(lineage_orders.region)`; `customer_seq` comes from `ROW_NUMBER()`. | +| `AGGREGATION` | The target value is produced by an expression that contains an aggregate function. It remains this type even when another expression wraps the aggregate. | `total_amount` comes from `SUM(lineage_orders.amount)`. | + +This example produces the following direct column lineage: + +| Target column | Source expression | Type | +| --- | --- | --- | +| `lineage_customer_summary.customer_id` | `lineage_orders.customer_id` | `IDENTITY` | +| `lineage_customer_summary.region_label` | `UPPER(lineage_orders.region)` | `TRANSFORMATION` | +| `lineage_customer_summary.total_amount` | `SUM(lineage_orders.amount)` | `AGGREGATION` | +| `lineage_customer_summary.customer_seq` | `ROW_NUMBER() OVER (ORDER BY lineage_orders.customer_id)` | `TRANSFORMATION` | +| `lineage_customer_summary.region_group` | `CASE WHEN UPPER(lineage_orders.region) = 'EAST' ... END` | `TRANSFORMATION` | + +#### Dataset indirect lineage + +Dataset indirect lineage records expressions that affect the complete result set without directly producing a target-column value. These dependencies are stored in an event-level map and can be applied to every target output column when needed. + +| Type | Meaning | Example | +| --- | --- | --- | +| `JOIN` | Join conditions determine which source rows can be combined. | `lineage_orders.customer_id = lineage_customers.customer_id`. | +| `FILTER` | `WHERE` or `HAVING` conditions determine which rows or groups enter the result. | `status = 'PAID'`, `customer_level = 'VIP'`, and `HAVING SUM(amount) >= 50`. | +| `GROUP_BY` | Grouping expressions determine the granularity of aggregate results. | Group by `customer_id` and `UPPER(region)`. | +| `SORT` | `ORDER BY` or TopN expressions determine result order. | Sort by `total_amount DESC`, which resolves to `SUM(amount)`. | + +This example produces the following dataset indirect lineage: + +| Type | Example source expression in the event | Effect | +| --- | --- | --- | +| `JOIN` | `lineage_orders.customer_id = lineage_customers.customer_id` | Determines how orders match customers. | +| `FILTER` | `status = 'PAID'`, `customer_level = 'VIP'`, and `SUM(amount) >= 50` | The `WHERE` and `HAVING` predicates jointly determine which rows and groups enter the result. | +| `GROUP_BY` | `lineage_orders.customer_id` and `UPPER(lineage_orders.region)` | Determine the aggregation granularity for each customer and region. | +| `SORT` | `total_amount DESC`, resolved to `SUM(lineage_orders.amount)` | Determines the order of the Insert query result. | + +#### Output indirect lineage + +Output indirect lineage is attached only to the affected target column and does not apply to the other output columns. + +| Type | Meaning | Example | +| --- | --- | --- | +| `WINDOW` | The `PARTITION BY` and `ORDER BY` inputs of a window function affect its output column. | `customer_seq` is produced by `ROW_NUMBER() OVER (ORDER BY customer_id)`, whose `WINDOW` dependency is `customer_id`. | +| `CONDITIONAL` | A `CASE` or `IF` condition, or the candidate expressions of `COALESCE`, affect its output column. | `region_group` is determined by `CASE WHEN region_label = 'EAST' ...`, whose `CONDITIONAL` dependency is that predicate. | + +The window or conditional expression itself still appears in direct column lineage. For example, the direct type of `customer_seq` is `TRANSFORMATION`, and it also has `WINDOW` indirect lineage. The direct type of `region_group` is also `TRANSFORMATION`, and it also has `CONDITIONAL` indirect lineage. + +This example produces the following output indirect lineage: + +| Target column | Type | Source expression in the event | Effect | +| --- | --- | --- | --- | +| `customer_seq` | `WINDOW` | `lineage_orders.customer_id` | Serves as the window ordering input to `ROW_NUMBER()` and affects only `customer_seq`. | +| `region_group` | `CONDITIONAL` | `UPPER(lineage_orders.region) = 'EAST'` | Serves as the `CASE WHEN` condition and affects only `region_group`. | + +The other three target columns have no output indirect lineage. Dataset-level `JOIN`, `FILTER`, `GROUP_BY`, and `SORT` dependencies still affect the complete result set. + +#### Query context + +Query context identifies who produced the lineage event, in which session, and through which DML statement. + +| Field | Example or meaning | +| --- | --- | +| Source Command | `InsertIntoTableCommand`, indicating that the event comes from an Insert. | +| Query ID and original SQL | The Query ID and the complete `INSERT INTO lineage_customer_summary ...` text. | +| User and client IP | For example, `lineage_user` and `192.0.2.10`. | +| Session database and catalog | `lineage_demo` and `internal` in this example. They describe session context and do not necessarily identify the target-table location. | +| Execution state | A successful DML statement normally records `OK`. | +| Timestamp and duration | The event creation time and the milliseconds from query start to event creation. The value is `-1` when unavailable. | +| External catalog properties | Non-sensitive properties from external catalogs referenced by the query. Passwords, keys, and hidden properties are removed. The map is empty when only the `internal` catalog is used. | + +For this example, the original SQL is the complex Insert above, the Source Command is `InsertIntoTableCommand`, the session database is `lineage_demo`, the session catalog is `internal`, and the successful state is `OK`. The Query ID, user, client IP, timestamp, and duration use values from the actual execution context. Because the example uses only the Internal Catalog, the external catalog properties map is empty. + +Confirm in the plugin's downstream system that the event was received. During initial deployment, also check `fe.log` for `Loaded lineage plugin`, plugin-specific errors, and queue-full warnings. + +![LineageInfo event model: target and source tables feed direct and indirect lineage, which are combined with query context before a plugin converts the event for a downstream system.](/images/data-lineage/lineage-event-model.svg) + +The extractor resolves CTE producer expressions and expands `UNION` branches before the event is delivered. A downstream plugin must convert Doris Java objects such as `TableIf`, `SlotReference`, and `Expression` to stable identifiers or its own event schema. + +### Event processing behavior + +`lineage_event_queue_size` sets the maximum number of lineage events waiting in each FE-local queue. It counts events, not bytes. When the queue is full, the new event is discarded and the DML continues normally. The worker invokes plugins serially, so a slow plugin delays later events and can cause drops under sustained load. Exceptions thrown by a plugin are logged and do not stop the worker or affect the DML. + +Plugin discovery and initialization occur only when FE starts. Dynamic reload and unload are not supported. + +## Configure and use + +### Prerequisites + +1. Prepare an external lineage plugin. Doris does not ship a built-in sink. For the SPI contract, JAR packaging, and a complete minimal plugin, see [Data Lineage Plugin Development](/community/developer-guide/data-lineage-plugin-development). +2. Copy the plugin JAR and any required third-party dependency JARs to every FE. +3. If the plugin sends events to an external governance service, make sure that service is reachable from every FE that can execute the DML. + +### Deploy the plugin + +`FE_HOME` denotes a single FE installation directory that contains `bin/`, `conf/`, and `lib/`. Use the following layout on every FE. The loader scans only direct child directories under `lineage/`, plus JARs in each plugin directory and its `lib/` directory. + +```text +$FE_HOME/plugins/ +└── lineage/ + └── example-lineage/ + ├── example-lineage.jar + └── lib/ + └── downstream-client.jar +``` + +Configure `fe.conf` on every FE. Replace `example-lineage` with the value returned by the plugin factory's `name()` method: + +```text +plugin_dir = /opt/apache-doris/fe/plugins +activate_lineage_plugin = example-lineage +lineage_event_queue_size = 50000 +``` + +| Configuration | Type and default | Required | Description | +| --- | --- | --- | --- | +| `plugin_dir` | String; `$FE_HOME/plugins` | No | Plugin root directory. The lineage loader scans its direct children under `lineage/`. Omit this setting when using the default directory. | +| `activate_lineage_plugin` | String array; empty | Explicit configuration recommended | Comma-separated Factory names to instantiate. An empty value disables name filtering and instantiates all discovered factories. | +| `lineage_event_queue_size` | Positive integer; `50000` | No | Maximum pending lineage events on the current FE. New events are discarded when the queue is full. | + +All three settings are configured in `$FE_HOME/conf/fe.conf` on each FE and apply only to that FE process. They are read only at FE startup, so restart the corresponding FE after changing them. + +#### Activate plugins + +Configure `activate_lineage_plugin` in `$FE_HOME/conf/fe.conf` on every FE. It controls which discovered factories are instantiated. It does not discover JARs or control queue capacity. Each name is case-sensitive and must exactly match `LineagePluginFactory.name()`. The `LineagePlugin.name()` implementation should return the same name. The plugin directory name is not used for matching. Separate multiple plugin names with commas. The FE configuration parser trims the spaces around each comma-separated item, but name matching remains case-sensitive: + +```text +activate_lineage_plugin = example-lineage,governance-lineage +``` + +At startup, FE discovers built-in and external factories, then creates instances only for the configured names. Each FE reads its own configuration, so every FE that can execute DML must use the same plugin set. Restart FE after a change. This setting cannot be changed dynamically through `ADMIN SET FRONTEND CONFIG`. + +:::caution Empty value behavior + +In the current implementation, an empty `activate_lineage_plugin` value does not disable lineage plugins. It skips name filtering and instantiates every discovered factory. Explicitly list the plugins to enable, and do not use an empty value as a disable switch. After a plugin is loaded, `eventFilter()` still decides whether it receives an event before extraction and before dispatch. + +::: + +#### Configure the event queue + +Configure `lineage_event_queue_size` in `$FE_HOME/conf/fe.conf` on every FE. It must be a positive integer. Each FE that can execute DML has an independent queue and must be configured separately. + +This is an FE startup setting and cannot be changed dynamically through `ADMIN SET FRONTEND CONFIG`. Restart the corresponding FE after changing it. Increasing the value increases pending-event capacity and FE memory usage. It does not add consumer threads or increase plugin processing throughput. When events are missing, first search `fe.log` for `the lineage event queue is full` and reduce latency in the plugin's `exec()` method. Then size the queue according to peak backlog and available FE memory. + +Restart every FE after changing other plugin settings or replacing plugin JARs. Plugin configuration and directories are read only at FE startup. + +## Operate and troubleshoot + +### No event after a write + +Common causes include an unsupported statement, a failed DML statement, no plugin loaded on the current FE, or `eventFilter()` returning `false`. + +Check in this order: + +1. Confirm that the statement is a successful `INSERT INTO ... SELECT`, `INSERT OVERWRITE TABLE ... SELECT`, or `CREATE TABLE AS SELECT`. A `VALUES`-only Insert does not generate an event. +2. Confirm that the FE that executed the DML has configured and loaded the plugin. Search `fe.log` for `Loaded lineage plugin`. +3. Confirm that the plugin's `eventFilter()` returns `true` in both the query thread and the worker thread. +4. Check plugin logs and the downstream service to confirm that processing did not fail outside FE. + +### Plugin not loaded + +Confirm that the plugin directory contains a JAR and that the JAR contains `META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory`. Each plugin directory can expose only one Factory. Factory names must be globally unique and must match `activate_lineage_plugin`. Restart the corresponding FE after correcting the directory, JAR, or configuration. + +### Missing events under high load + +Search `fe.log` for `the lineage event queue is full` and plugin exceptions. If the queue is full, first reduce synchronous blocking in `exec()`, and make downstream delivery idempotent with bounded retries. Then evaluate whether to increase `lineage_event_queue_size`. Before increasing it, account for FE memory because event sizes vary by SQL statement. + +Also confirm that every FE that can execute DML has the same plugin and configuration. Each FE uses an independent local queue; plugin and queue state are not synchronized across FEs. + +### Plugin update did not take effect + +Plugin directories are scanned only when FE starts. Dynamic reload and unload are not supported. Restart every relevant FE after replacing plugin JARs, dependencies, or configuration, and check `Loaded lineage plugin` again. + +Use the Query ID and target table to construct downstream event identifiers. This allows the governance system to deduplicate events even when the plugin implements retries outside the framework. + +## Further reading + +- [Data Lineage Plugin Development](/community/developer-guide/data-lineage-plugin-development) diff --git a/i18n/zh-CN/docusaurus-plugin-content-docs-community/current.json b/i18n/zh-CN/docusaurus-plugin-content-docs-community/current.json index 3b65ed24390cd..d18af08705553 100644 --- a/i18n/zh-CN/docusaurus-plugin-content-docs-community/current.json +++ b/i18n/zh-CN/docusaurus-plugin-content-docs-community/current.json @@ -43,6 +43,10 @@ "message": "数据源扩展开发", "description": "The label for category Data Source Extension in sidebar community" }, + "sidebar.community.category.Plugin Development": { + "message": "插件开发", + "description": "The label for category Plugin Development in sidebar community" + }, "sidebar.community.category.Data Format Reference": { "message": "数据格式参考", "description": "The label for category Data Format Reference in sidebar community" diff --git a/i18n/zh-CN/docusaurus-plugin-content-docs-community/current/developer-guide/data-lineage-plugin-development.md b/i18n/zh-CN/docusaurus-plugin-content-docs-community/current/developer-guide/data-lineage-plugin-development.md new file mode 100644 index 0000000000000..d7b7262d4b773 --- /dev/null +++ b/i18n/zh-CN/docusaurus-plugin-content-docs-community/current/developer-guide/data-lineage-plugin-development.md @@ -0,0 +1,744 @@ +--- +title: 数据血缘插件开发 +language: zh-CN +description: 介绍如何基于 Apache Doris 4.0.6 源码逐步开发、编译、打包、部署和验证数据血缘插件，说明事件模型、目录路径、FE 配置、日志验证、可靠性处理和故障排查，并介绍对接 OpenMetadata、OpenLineage 等外部治理组件的扩展方式。 +keywords: + - Apache Doris + - 数据血缘 + - LineagePlugin + - LineagePluginFactory + - ServiceLoader + - 插件开发 +--- + + + + + + +本文介绍如何为 Apache Doris 开发外部数据血缘插件。插件会在受支持的 DML 成功执行后接收 `LineageInfo` 事件，再将其转换并写入日志、消息系统或外部元数据治理平台。 + +> 适用版本：Apache Doris 社区开源版 4.0.6 及以后版本。插件必须针对 FE 实际部署的 Doris 社区版本或源码修订版本编译。 + +## 范围和生命周期 + +血缘框架是 FE 侧的 SPI。它不保存血缘数据、不提供查询 API，也不渲染血缘图。插件负责将 `LineageInfo` 中的内存 Java 对象转换为下游系统要求的格式和传输协议。 + +当前实现仅为成功执行的 `INSERT INTO ... SELECT`、`INSERT OVERWRITE TABLE ... SELECT` 和 `CREATE TABLE AS SELECT` 生成事件。仅包含 `VALUES` 的写入和写入目标为 `__internal_schema` 的操作会被跳过。`SELECT`、`UPDATE`、`DELETE` 以及各类导入任务不在该框架的覆盖范围内。部分 `UPDATE` 和 `DELETE` 执行路径会在内部复用 Insert 命令，但原始命令类型校验会阻止这些命令提交血缘事件。 + +![数据血缘采集架构：受支持的 DML 成功后，由 Nereids 分析并提取为 LineageInfo，经 FE 队列和插件投递到外部治理系统。](/images/data-lineage/lineage-architecture-zh-CN.svg) + +`eventFilter()` 会在 DML 查询路径上、血缘提取前调用一次，也会在工作线程分发前再次调用。插件不需要接收事件时应返回 `false`。该方法会被查询线程和工作线程并发调用，因此必须线程安全。`exec()` 只由一个工作线程调用，但可能与 `eventFilter()` 并发执行。 + +FE 启动时先通过 ServiceLoader 发现 `LineagePluginFactory`，再调用 Factory 的 `create(context)` 创建插件，随后调用插件的 `initialize(context)`。`LineagePluginFactory` 默认会把 `create(context)` 委托给无参数的 `create()`，因此本文的最小示例只需实现 `create()`。`PluginContext` 通过 `getProperties()` 暴露不可变属性 Map，FE 将插件名称和目录分别存入 `plugin.name` 和 `plugin.path`。插件需要通过 `context.getProperties().get("plugin.name")` 和 `context.getProperties().get("plugin.path")` 读取这两个属性。生产插件如需读取插件目录内的配置文件，可以覆盖 `initialize(context)`，根据 `plugin.path` 属性定位文件并完成资源初始化。 + +## 血缘事件模型 + +`LineageInfo` 包含目标表、目标输出列、源表、直接血缘、间接血缘和 `LineageContext`。 + +| 部分 | 含义 | +| --- | --- | +| 目标 | 写入的目标表和目标输出列。 | +| 源表 | 已分析逻辑计划中引用的表。 | +| 直接血缘 | 每个输出 Slot 到源表达式的映射，类型为 `IDENTITY`、`TRANSFORMATION` 或 `AGGREGATION`。 | +| 数据集级间接血缘 | 影响整个结果集的表达式：`JOIN`、`FILTER`、`GROUP_BY` 和 `SORT`。 | +| 输出列级间接血缘 | 仅影响特定输出列的依赖：`WINDOW` 和 `CONDITIONAL`。 | +| 查询上下文 | Query ID、SQL 文本、用户、客户端 IP、会话库和 Catalog、执行状态、时间戳、耗时及已脱敏的外部 Catalog 属性。 | + +直接血缘描述输出值的产生方式：纯源列引用为 `IDENTITY`，例如 `target.customer_id <- source.customer_id`；不包含聚合函数的计算、函数、窗口或条件表达式为 `TRANSFORMATION`，例如 `UPPER(source.region)`；包含聚合函数的表达式为 `AGGREGATION`，例如 `SUM(source.amount)`。间接血缘记录影响输出、但不直接成为输出值的表达式，例如 Join Key、`WHERE`/`HAVING` 条件、分组键和排序键。`WINDOW` 记录窗口的分区和排序输入，`CONDITIONAL` 记录 `CASE`、`IF` 或 `COALESCE` 对特定输出列的影响。 + +### 血缘事件模型示例 + +以下示例通过一个源表和一个目标表覆盖三类直接血缘、数据集级间接血缘以及输出列级间接血缘。示例需要当前用户具有创建数据库、创建表、写入和查询权限。 + +```sql +CREATE DATABASE IF NOT EXISTS lineage_demo; +USE lineage_demo; + +CREATE TABLE lineage_source ( + region VARCHAR(32), + amount DECIMAL(18, 2), + status VARCHAR(16) +) +DUPLICATE KEY(region) +DISTRIBUTED BY HASH(region) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_target ( + region VARCHAR(32), + total_amount DECIMAL(18, 2), + region_seq BIGINT, + region_group VARCHAR(16) +) +DUPLICATE KEY(region) +DISTRIBUTED BY HASH(region) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +INSERT INTO lineage_source VALUES + ('east', 100.00, 'PAID'), + ('east', 30.00, 'CANCELLED'), + ('west', 80.00, 'PAID'); + +INSERT INTO lineage_target +SELECT + region, + SUM(amount) AS total_amount, + ROW_NUMBER() OVER (ORDER BY region) AS region_seq, + CASE + WHEN region = 'east' THEN 'CORE' + ELSE 'OTHER' + END AS region_group +FROM lineage_source +WHERE status = 'PAID' +GROUP BY region +ORDER BY SUM(amount) DESC; + +SELECT region, total_amount, region_seq, region_group +FROM lineage_target +ORDER BY region; +``` + +查询结果如下： + +```text ++--------+--------------+------------+--------------+ +| region | total_amount | region_seq | region_group | ++--------+--------------+------------+--------------+ +| east | 100.00 | 1 | CORE | +| west | 80.00 | 2 | OTHER | ++--------+--------------+------------+--------------+ +``` + +仅包含 `VALUES` 的第一个 Insert 不会生成事件。`INSERT INTO lineage_target ... SELECT` 成功后生成的事件包含以下关键关系。 + +| 血缘层级 | 该示例中的内容 | +| --- | --- | +| 表级血缘 | `internal.lineage_demo.lineage_source` -> `internal.lineage_demo.lineage_target`。 | +| 列级直接血缘 | `region` 为 `IDENTITY`；`total_amount` 为 `AGGREGATION`；`region_seq` 和 `region_group` 为 `TRANSFORMATION`。 | +| 数据集级间接血缘 | `FILTER` 为 `status = 'PAID'`；`GROUP_BY` 为 `region`；`SORT` 为 `SUM(amount) DESC`。 | +| 输出列级间接血缘 | `region_seq` 的 `WINDOW` 依赖为 `region`；`region_group` 的 `CONDITIONAL` 依赖为 `region = 'east'`。 | +| 查询上下文 | 包含实际 Query ID、上述 Insert SQL、执行用户和客户端 IP；会话数据库为 `lineage_demo`，Internal Catalog 为 `internal`，执行状态为 `OK`。 | + +该示例没有 Join，因此不会产生 `JOIN` 间接血缘。带 Join 的查询会把 Join 条件记录为数据集级 `JOIN` 依赖。 + +事件到达插件前，提取器会解析 CTE 生产者表达式，并展开 `UNION` 的各分支。插件仍应将 `Expression`、`SlotReference` 和 `TableIf` 视为 Doris 内部 Java 对象，在发送到 FE 外部前转换为稳定的名称或下游系统格式。 + +## 插件 API 参考 + +数据血缘 SPI 由 FE 启动时发现的 Factory、Factory 创建的插件实例、不可变运行时上下文，以及传给插件的 `LineageInfo` 事件组成。下表汇总插件实现直接使用的扩展方法。 + +### API 总览 + +| 方法 | 输入 | 返回类型 | 作用 | +| --- | --- | --- | --- | +| `LineagePluginFactory.name()` | 无 | `String` | 返回 `activate_lineage_plugin` 使用的唯一 Factory 名称。 | +| `LineagePluginFactory.create()` | 无 | `LineagePlugin` | 创建插件实例，Factory 不读取运行时上下文。 | +| `LineagePluginFactory.create(PluginContext)` | `PluginContext` | `LineagePlugin` | 使用运行时上下文创建插件实例；默认实现委托给 `create()`。 | +| `LineagePlugin.name()` | 无 | `String` | 返回插件名称，应与 Factory 名称一致。 | +| `LineagePlugin.eventFilter()` | 无 | `boolean` | 表示插件当前是否接收血缘事件。 | +| `LineagePlugin.exec(LineageInfo)` | `LineageInfo` | `boolean` | 处理一条血缘事件。 | +| `Plugin.initialize(PluginContext)` | `PluginContext` | `void` | Factory 创建插件后初始化资源；默认实现不执行操作。 | +| `Plugin.close()` | 无 | `void` | 用于释放插件资源的生命周期 Hook；默认实现不执行操作。 | +| `PluginContext.getProperties()` | 无 | `Map` | 返回不可变的运行时属性 Map。 | + +### `LineagePluginFactory` + +`LineagePluginFactory` 是 ServiceLoader 的入口。FE 发现 Factory 后，根据 `name()` 的返回值进行筛选，然后创建一个插件实例。 + +```java +String name(); +LineagePlugin create(); +default LineagePlugin create(PluginContext context); +``` + +#### `name()` + +- **输入：** 无。 +- **返回：** `String`，区分大小写的唯一 Factory 标识。 +- **行为：** 该值与 `activate_lineage_plugin` 中的每一项进行匹配。本文示例返回 `example-lineage`。 + +#### `create()` + +- **输入：** 无。 +- **返回：** 新的 `LineagePlugin` 实例。 +- **行为：** 插件构造不需要运行时上下文时实现该方法。默认的 `create(PluginContext)` 会委托给它。 + +#### `create(PluginContext context)` + +- **输入：** `PluginContext context`，FE 为当前插件目录准备的不可变运行时上下文。 +- **返回：** 新的 `LineagePlugin` 实例。 +- **行为：** 只有 Factory 构造插件时需要读取运行时属性才覆盖该方法。创建完成后，FE 会对返回的插件调用 `initialize(context)`。 + +创建插件时返回 `null` 或抛出异常会导致该插件无法加载。插件发现发生在 FE 启动阶段，因此不要在 Factory 方法中执行耗时的下游调用。 + +### `LineagePlugin` + +`LineagePlugin` 接收血缘事件，并从 `Plugin` 继承 `initialize()` 和 `close()` 生命周期方法。 + +```java +String name(); +boolean eventFilter(); +boolean exec(LineageInfo lineageInfo); +default void initialize(PluginContext context); +default void close(); +``` + +#### `name()` + +- **输入：** 无。 +- **返回：** `String`，插件标识。 +- **行为：** 应返回与 `LineagePluginFactory.name()` 相同的稳定标识，使激活配置、日志和运维配置使用同一个名称。 + +#### `eventFilter()` + +- **输入：** 无。 +- **返回：** `boolean`；`true` 表示插件接收血缘事件，`false` 表示 FE 跳过该插件。 +- **行为：** FE 会在查询线程提取血缘前调用该方法，并在血缘工作线程分发事件前再次调用。该方法必须线程安全且快速返回。如果提取前所有已加载插件都返回 `false`，FE 会跳过该语句的血缘提取。 + +#### `exec(LineageInfo lineageInfo)` + +- **输入：** `LineageInfo lineageInfo`，受支持 DML 成功后提取的一条血缘事件。 +- **返回：** `boolean`；`true` 表示插件报告处理成功，`false` 表示插件报告处理失败。 +- **行为：** 单个血缘工作线程调用 `exec()`，但它可能与查询线程中的 `eventFilter()` 并发执行。当前框架不会在 `exec()` 返回 `false` 后重试、重新入队或修改 DML 状态。插件抛出的异常会被记录，后续插件或事件仍会继续处理。 + +#### `initialize(PluginContext context)` + +- **输入：** `PluginContext context`，与传给 Factory 的运行时上下文相同。 +- **返回：** `void`。 +- **行为：** FE 创建插件后调用该方法。插件可以覆盖它，根据 `plugin.path` 读取配置、创建客户端或分配其他资源；抛出异常会导致插件无法加载。 + +#### `close()` + +- **输入：** 无。 +- **返回：** `void`。 +- **行为：** 这个继承的生命周期 Hook 用于释放资源。当前血缘处理器没有动态卸载或重新加载流程，因此插件不能依赖替换 JAR 时一定会调用 `close()`；替换 JAR 后仍需重启 FE。 + +### `PluginContext` + +```java +Map getProperties(); +``` + +`getProperties()` 没有输入，返回不可变的 `Map`。血缘加载器提供以下属性： + +| 属性 | 类型 | 含义 | +| --- | --- | --- | +| `plugin.name` | `String` | 已发现 Factory 返回的名称。 | +| `plugin.path` | `String` | 外部插件目录的绝对路径。 | + +通过 `context.getProperties().get("plugin.name")` 和 `context.getProperties().get("plugin.path")` 读取这两个属性。该 Map 不可修改。插件可以使用 `plugin.path` 定位自己的配置文件，但不能假设插件目录名等于 Factory 名称。 + +### `LineageInfo` + +`LineageInfo` 是 `exec()` 的输入类型。插件应通过以下方法读取事件，并将 Doris 对象转换为下游事件格式。 + +| 方法 | 返回类型 | 含义 | +| --- | --- | --- | +| `getTargetTable()` | `TableIf` | 写入的目标表。 | +| `getTargetColumns()` | `List` | 按写入顺序排列的目标输出列。 | +| `getTableLineageSet()` | `Set` | 已分析计划中引用的源表。 | +| `getDirectLineageMap()` | `Map>` | 每个输出 Slot 的直接源表达式。 | +| `getDatasetIndirectLineageMap()` | `Multimap` | 数据集级 `JOIN`、`FILTER`、`GROUP_BY` 和 `SORT` 依赖。 | +| `getInDirectLineageMapByDataset()` | `Map>` | 将数据集级依赖应用到每个输出 Slot 后得到的视图，不包含 `WINDOW` 和 `CONDITIONAL`。 | +| `getOutputIndirectLineageMap()` | `Map>` | 每个输出列的 `WINDOW` 和 `CONDITIONAL` 依赖。 | +| `getContext()` | `LineageContext` | 当前事件的查询和执行元数据。 | + +`LineageInfo` 还提供 FE 提取器使用的 setter 和 `add*` 方法。sink 插件通常只通过上述 getter 消费对象，不应修改事件。返回的 `TableIf`、`Slot`、`SlotReference` 和 `Expression` 是 Doris 运行时对象，不是稳定的传输协议标识。 + +### `LineageContext` + +`LineageContext` 由 `LineageInfo.getContext()` 返回，以下 getter 都没有输入。 + +| 方法 | 返回类型 | 含义 | +| --- | --- | --- | +| `getSourceCommand()` | `Class` | 产生事件的命令类型，例如 `InsertIntoTableCommand`。 | +| `getQueryId()` | `String` | Query ID。 | +| `getQueryText()` | `String` | 原始 DML 文本。 | +| `getUser()` | `String` | 执行用户。 | +| `getClientIp()` | `String` | 客户端 IP。 | +| `getState()` | `String` | 查询执行状态。 | +| `getDatabase()` | `String` | 会话数据库，不一定是目标表所在数据库。 | +| `getCatalog()` | `String` | 会话 Catalog，不一定是目标表所在 Catalog。 | +| `getTimestampMs()` | `long` | 事件时间戳，单位为毫秒；不可用时为 `-1`。 | +| `getDurationMs()` | `long` | 查询耗时，单位为毫秒；不可用时为 `-1`。 | +| `getExternalCatalogProperties()` | `Map>` | 查询引用的外部 Catalog 的已脱敏属性。 | + +原始 SQL、用户、客户端 IP、表达式和外部 Catalog 属性可能包含敏感数据。生产插件在导出前应按自身安全策略实施访问控制、脱敏、长度限制和保留策略。 + +## 开发插件 + +本节以 `github.com/apache/doris` 官方社区开源仓库的 `4.0.6` 标签为例，在源码内新增一个 `example-lineage` Maven 子模块。插件收到事件后只写入 FE 日志，不启动 HTTP 服务，也不依赖外部治理平台。该实现用于演示完整的插件开发和加载流程；实际插件可以将 `LineageInfo` 转换为第三方系统需要的格式，调用 OpenMetadata 等元数据治理平台的接口，或发送符合 OpenLineage 规范的血缘事件。以下所有源码路径都相对于同一个 Doris 源码根目录。 + +### 第 1 步：准备 Doris 源码 + +从 Apache Doris 官方社区开源仓库检出与目标 FE 完全一致的公开版本，并用 `DORIS_SOURCE` 记录源码根目录： + +```shell +git clone https://github.com/apache/doris.git +export DORIS_SOURCE="$(pwd)/doris" +cd "${DORIS_SOURCE}" +git checkout 4.0.6 +``` + +后文中的 `$DORIS_SOURCE/fe_plugins/...` 都指向这个源码目录。不要用 FE 安装目录替代源码目录，也不要使用其他版本的 FE JAR 编译插件。 + +### 第 2 步：创建 Maven 模块 + +在 Doris 源码根目录执行： + +```shell +cd "${DORIS_SOURCE}" +mkdir -p fe_plugins/lineage/example-lineage/src/main/java/org/apache/doris/plugin/lineage/example +mkdir -p fe_plugins/lineage/example-lineage/src/main/resources/META-INF/services +``` + +完成本节后，新增文件应形成以下结构： + +```text +$DORIS_SOURCE/ +└── fe_plugins/ + ├── pom.xml # 已有文件，需要增加 lineage 模块 + └── lineage/ + ├── pom.xml + └── example-lineage/ + ├── pom.xml + └── src/main/ + ├── java/org/apache/doris/plugin/lineage/example/ + │ ├── ExampleLineagePluginFactory.java + │ └── LoggingLineagePlugin.java + └── resources/META-INF/services/ + └── org.apache.doris.nereids.lineage.LineagePluginFactory +``` + +编辑已有文件 `$DORIS_SOURCE/fe_plugins/pom.xml`，在其 `` 中增加 `lineage`。保留原有模块，不要用下面片段覆盖整个文件： + +```xml + + + lineage + +``` + +创建 `$DORIS_SOURCE/fe_plugins/lineage/pom.xml`： + +```xml + + + + 4.0.0 + + org.apache.doris + fe-plugins + 1.0-SNAPSHOT + ../pom.xml + + lineage + pom + + example-lineage + + +``` + +创建 `$DORIS_SOURCE/fe_plugins/lineage/example-lineage/pom.xml`： + +```xml + + + + 4.0.0 + + org.apache.doris + lineage + 1.0-SNAPSHOT + ../pom.xml + + example-lineage + jar + + + org.apache.doris + fe-core + ${doris.version} + provided + + + org.apache.doris + fe-extension-spi + ${doris.version} + provided + + + org.apache.logging.log4j + log4j-api + provided + + + + example-lineage + + +``` + +所有依赖都使用 `provided`，因为这些 API 由 FE 提供。插件 JAR 中不能再次打入 Doris 或 Log4j 类。 + +### 第 3 步：实现 Factory + +创建 `$DORIS_SOURCE/fe_plugins/lineage/example-lineage/src/main/java/org/apache/doris/plugin/lineage/example/ExampleLineagePluginFactory.java`： + +```java +// 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. + +package org.apache.doris.plugin.lineage.example; + +import org.apache.doris.nereids.lineage.LineagePlugin; +import org.apache.doris.nereids.lineage.LineagePluginFactory; + +/** Factory discovered by ServiceLoader at FE startup. */ +public class ExampleLineagePluginFactory implements LineagePluginFactory { + @Override + public String name() { + return "example-lineage"; + } + + @Override + public LineagePlugin create() { + return new LoggingLineagePlugin(); + } +} +``` + +`name()` 返回的 `example-lineage` 是插件的唯一标识，后续 `activate_lineage_plugin` 必须使用完全相同的名称。 + +### 第 4 步：实现日志插件 + +创建 `$DORIS_SOURCE/fe_plugins/lineage/example-lineage/src/main/java/org/apache/doris/plugin/lineage/example/LoggingLineagePlugin.java`： + +```java +// 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. + +package org.apache.doris.plugin.lineage.example; + +import org.apache.doris.catalog.TableIf; +import org.apache.doris.nereids.lineage.LineageContext; +import org.apache.doris.nereids.lineage.LineageInfo; +import org.apache.doris.nereids.lineage.LineagePlugin; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.Objects; +import java.util.stream.Collectors; + +/** Minimal lineage plugin that writes one summary line for each event. */ +public class LoggingLineagePlugin implements LineagePlugin { + private static final Logger LOG = LogManager.getLogger(LoggingLineagePlugin.class); + private static final String PLUGIN_NAME = "example-lineage"; + + @Override + public String name() { + return PLUGIN_NAME; + } + + @Override + public boolean eventFilter() { + return true; + } + + @Override + public boolean exec(LineageInfo lineageInfo) { + if (lineageInfo == null) { + return false; + } + + LineageContext context = lineageInfo.getContext(); + LOG.info("LINEAGE_EVENT queryId={} command={} user={} clientIp={} " + + "catalog={} database={} state={} durationMs={} target={} sources={} " + + "direct={} datasetIndirect={} outputIndirect={}", + value(context == null ? null : context.getQueryId()), + command(context), + value(context == null ? null : context.getUser()), + value(context == null ? null : context.getClientIp()), + value(context == null ? null : context.getCatalog()), + value(context == null ? null : context.getDatabase()), + value(context == null ? null : context.getState()), + context == null ? -1L : context.getDurationMs(), + tableName(lineageInfo.getTargetTable()), + sourceTables(lineageInfo), + lineageInfo.getDirectLineageMap(), + lineageInfo.getDatasetIndirectLineageMap(), + lineageInfo.getOutputIndirectLineageMap()); + return true; + } + + private String sourceTables(LineageInfo lineageInfo) { + if (lineageInfo.getTableLineageSet() == null) { + return "[]"; + } + return lineageInfo.getTableLineageSet().stream() + .filter(Objects::nonNull) + .map(TableIf::getNameWithFullQualifiers) + .sorted() + .collect(Collectors.joining(",", "[", "]")); + } + + private String tableName(TableIf table) { + return table == null ? "" : table.getNameWithFullQualifiers(); + } + + private String command(LineageContext context) { + return context == null || context.getSourceCommand() == null + ? "" : context.getSourceCommand().getSimpleName(); + } + + private String value(String value) { + return value == null ? "" : value; + } +} +``` + +该示例不打印原始 SQL、外部 Catalog 属性或认证信息。直接和间接血缘表达式仍可能包含 SQL 字面量，因此该实现只适合开发验证；生产插件应按自身安全规范对表达式脱敏、限制单条日志长度，并决定是否记录用户和客户端 IP。 + +### 第 5 步：注册 Factory + +创建 `$DORIS_SOURCE/fe_plugins/lineage/example-lineage/src/main/resources/META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory`，文件中只写一行 Factory 类名： + +```text +org.apache.doris.plugin.lineage.example.ExampleLineagePluginFactory +``` + +每个外部插件目录只能暴露一个 `LineagePluginFactory`。Factory 名称还必须在所有已加载插件中保持唯一。 + +### 第 6 步：构建插件 + +首次在该源码目录中构建插件前，需要先准备 Doris 的源码编译环境。根据操作系统完成 [Linux 编译环境准备](/community/source-install/compilation-linux)或 [macOS 编译环境准备](/community/source-install/compilation-mac)，再执行一次 FE 构建。以下环境变量只关闭血缘插件编译不需要的 UI、Hive UDF 和 BE Java 扩展，不会跳过 FE： + +```shell +cd "${DORIS_SOURCE}" +DISABLE_BUILD_UI=ON \ +DISABLE_BE_JAVA_EXTENSIONS=ON \ +DISABLE_BUILD_HIVE_UDF=ON \ +./build.sh --fe +``` + +该步骤会生成 Thrift 和 Protobuf 源码并构建 FE。不能在全新检出的源码树中跳过该步骤直接执行下面的 Maven 命令；否则会因缺少 `org.apache.doris.thrift` 等生成类而编译失败。同一源码版本已成功执行过 `./build.sh --fe` 时不需要重复执行。 + +接着将插件依赖的 FE 模块安装到本地 Maven 仓库，再构建插件子模块： + +```shell +cd "${DORIS_SOURCE}" +mvn -f fe/pom.xml install \ + -pl fe-common,fe-extension-spi,fe-extension-loader,fe-core \ + -DskipTests -Dcheckstyle.skip=true +mvn -f fe_plugins/pom.xml -pl lineage/example-lineage -am clean package \ + -DskipTests -Dcheckstyle.skip=true +``` + +构建成功后生成： + +```text +$DORIS_SOURCE/fe_plugins/lineage/example-lineage/target/example-lineage.jar +``` + +检查 SPI 文件和 JAR 内容： + +```shell +PLUGIN_JAR="${DORIS_SOURCE}/fe_plugins/lineage/example-lineage/target/example-lineage.jar" +unzip -p "${PLUGIN_JAR}" \ + META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory +if jar tf "${PLUGIN_JAR}" | grep -qE '^org/apache/doris/(nereids|extension)/'; then + echo "ERROR: Doris API classes must not be packaged in the plugin JAR" >&2 + exit 1 +fi +``` + +第一条命令应输出 `org.apache.doris.plugin.lineage.example.ExampleLineagePluginFactory`。后续检查在未打入 Doris API 时不输出内容并返回 0。 + +### 第 7 步：部署和激活插件 + +`FE_HOME` 在本节表示单个 FE 的安装目录，其中包含 `bin/`、`conf/` 和 `lib/`，不是前面的 Doris 源码目录。`start_fe.sh` 会将该目录设置为 FE 进程的 `DORIS_HOME`。将 JAR 复制到每个 FE： + +```shell +export FE_HOME=/opt/apache-doris/fe +mkdir -p "${FE_HOME}/plugins/lineage/example-lineage" +cp "${DORIS_SOURCE}/fe_plugins/lineage/example-lineage/target/example-lineage.jar" \ + "${FE_HOME}/plugins/lineage/example-lineage/" +``` + +插件目录及 JAR 必须对运行 FE 的操作系统用户可读。 + +部署后的目录为： + +```text +$FE_HOME/ +└── plugins/ + └── lineage/ + └── example-lineage/ + └── example-lineage.jar +``` + +编辑每个 FE 的 `$FE_HOME/conf/fe.conf`： + +```text +plugin_dir = /opt/apache-doris/fe/plugins +activate_lineage_plugin = example-lineage +lineage_event_queue_size = 50000 +``` + +| 配置 | 类型和默认值 | 是否必填 | 作用 | +| --- | --- | --- | --- | +| `plugin_dir` | String；`$FE_HOME/plugins` | 否 | 插件根目录。血缘加载器扫描其 `lineage/` 直接子目录。使用默认目录时可以省略该配置。 | +| `activate_lineage_plugin` | String 数组；空 | 建议显式填写 | 需要实例化的 Factory 名称，以英文逗号分隔。空值会实例化全部已发现的 Factory。 | +| `lineage_event_queue_size` | 正整数；`50000` | 否 | 当前 FE 等待工作线程处理的最大事件数，超过后新事件将被丢弃。 | + +这些参数作用于当前 FE 进程，只在启动时读取。所有可能执行 DML 的 FE 都需要部署并配置插件；修改配置或替换 JAR 后必须重启对应 FE，不支持通过 `ADMIN SET FRONTEND CONFIG` 动态生效。 + +## 验证插件 + +重启 FE 后，先确认插件加载成功。如果修改过 FE 日志目录，请将下面的 `FE_LOG` 替换为实际 `fe.log` 路径： + +```shell +export FE_LOG="${FE_HOME}/log/fe.log" +grep 'Loaded lineage plugin: example-lineage' "${FE_LOG}" | tail -1 +``` + +然后执行前文的[血缘事件模型示例](#血缘事件模型示例)，再查询最新事件： + +```shell +grep 'LINEAGE_EVENT' "${FE_LOG}" | tail -1 +``` + +日志应包含实际 Query ID、`InsertIntoTableCommand`、目标表 `lineage_target`、源表 `lineage_source`，以及 `direct`、`datasetIndirect` 和 `outputIndirect` 三部分。例如： + +```text +LINEAGE_EVENT queryId= command=InsertIntoTableCommand ... +target=internal.lineage_demo.lineage_target +sources=[internal.lineage_demo.lineage_source] direct={...} +datasetIndirect={...} outputIndirect={...} +``` + +实际日志是一行；上面为便于阅读进行了换行。`INSERT INTO lineage_source VALUES ...` 不生成事件，只有 `INSERT INTO lineage_target ... SELECT` 会产生该日志。 + +验证完成且不再需要示例数据时，可以执行： + +```sql +DROP DATABASE lineage_demo; +``` + +## 可靠性和排障 + +| 情况 | 框架行为 | 插件建议 | +| --- | --- | --- | +| 队列已满 | 丢弃新事件并记录 warning，DML 结果不受影响。 | 保持 `exec()` 快速执行；在 FE 关键路径外批量或缓冲下游写入。 | +| `exec()` 抛出异常 | 工作线程记录异常后继续处理下一个插件或事件。 | 捕获可预期的下游异常，进行有界重试，并暴露监控指标。 | +| `exec()` 返回 `false` | 当前框架忽略返回值，不会重试、重新入队或改变 DML 状态。 | 插件必须自行记录失败并实现有界重试；不要依赖返回值触发恢复。 | +| 插件执行缓慢 | 单个工作线程串行分发事件，慢插件会延迟后续所有事件。 | 使用有界内部队列和独立发送线程池处理慢速下游。 | +| Factory 名称重复 | 保留第一个发现的 Factory，跳过重复项。 | 使用全局唯一的名称。 | +| 启动后替换 JAR | 加载器不监听目录变化，也不支持 reload 或 unload。 | 替换 JAR 或依赖后重启 FE。 | + +### 根据 FE 日志定位问题 + +| 日志关键字 | 常见原因 | 处理方法 | +| --- | --- | --- | +| `Skip lineage plugin directory due to load failure` | 插件目录中没有 JAR、SPI 文件缺失、暴露了多个 Factory，或依赖加载失败。 | 检查插件目录和 JAR 内容，确认 SPI 文件中只有一个 Factory 类名，并检查异常堆栈中的 `stage` 和 `message`。 | +| `Skip lineage plugin not in activate_lineage_plugin` | Factory 已发现，但其 `name()` 不在激活列表中。 | 核对 Factory 名称和 `activate_lineage_plugin`，名称区分大小写。 | +| `Failed to create/initialize lineage plugin` | `create(context)` 或 `initialize(context)` 抛出异常。 | 根据异常堆栈检查初始化逻辑、配置文件和目录读取权限。 | +| `the lineage event queue is full` | 插件消费速度低于事件产生速度，当前 FE 的队列已满。 | 优先缩短 `exec()` 执行时间并排查下游阻塞，再评估是否增大 `lineage_event_queue_size`。 | + +### 停用和回滚插件 + +停用插件时，先停止对应 FE，将插件目录移出 `plugin_dir/lineage/`，再从 `activate_lineage_plugin` 中移除 `example-lineage`，最后启动 FE。不要只清空 `activate_lineage_plugin` 而保留 JAR，因为空列表表示激活所有已发现的血缘插件。 + +```shell +mkdir -p "${FE_HOME}/plugins-disabled" +mv "${FE_HOME}/plugins/lineage/example-lineage" \ + "${FE_HOME}/plugins-disabled/" +``` + +启动 FE 后，确认日志中没有新的 `Loaded lineage plugin: example-lineage`。需要恢复时，将目录移回 `${FE_HOME}/plugins/lineage/`，恢复 `activate_lineage_plugin` 配置并再次重启 FE。 + +血缘投递采用尽力而为方式，不与 DML 事务绑定。DML 成功不代表事件已成功投递到外部治理系统。下游协议应以 Query ID 和目标表等信息构造幂等事件标识。 diff --git a/i18n/zh-CN/docusaurus-plugin-content-docs/current/data-governance/data-lineage.md b/i18n/zh-CN/docusaurus-plugin-content-docs/current/data-governance/data-lineage.md new file mode 100644 index 0000000000000..6848be7bae51f --- /dev/null +++ b/i18n/zh-CN/docusaurus-plugin-content-docs/current/data-governance/data-lineage.md @@ -0,0 +1,336 @@ +--- +{ + "title": "数据血缘原理与使用", + "sidebar_label": "数据血缘", + "language": "zh-CN", + "description": "介绍 Apache Doris 数据血缘的采集原理、事件内容、支持范围、插件部署、FE 配置、队列行为、运维排障和完整 SQL 示例，帮助用户理解并验证表级、列级以及直接和间接血缘关系。", + "keywords": ["Apache Doris", "数据血缘", "数据治理", "LineagePlugin", "列级血缘"] +} +--- + + + + +数据血缘功能可以从 Doris 支持的 DML 语句中提取表级和列级依赖，并通过 `LineagePlugin` 发送到外部治理系统。Doris 是血缘生产者，不保存事件、不提供血缘查询 API，也不提供血缘可视化。 + +> 该功能在 Dev 版本中可用，首次发布于 Apache Doris 4.0.6。 + +## 能力和限制 + +框架只会在以下语句成功后生成事件： + +| 支持的语句 | 事件行为 | +| --- | --- | +| `INSERT INTO ... SELECT` | Insert 成功后生成一个事件。 | +| `INSERT OVERWRITE TABLE ... SELECT` | Overwrite 成功后生成一个事件。 | +| `CREATE TABLE AS SELECT` | 内部 Insert 成功后生成一个事件。 | + +当前实现不会为 `SELECT`、`UPDATE`、`DELETE`、导入任务、仅包含 `VALUES` 的 Insert，以及目标表为 `__internal_schema` 的写入生成事件。部分 `UPDATE` 和 `DELETE` 执行路径会在内部复用 Insert 命令，但原始命令类型校验会阻止这些命令提交血缘事件。 + +:::caution 投递保证 + +血缘投递是异步且尽力而为的。DML 成功后，即使事件采集或插件投递失败，DML 也不会回滚。队列满时可能丢弃事件，Doris 不会重试或持久化这些事件。 + +::: + +## 工作原理 + +### 采集流程 + +对于受支持的语句，Doris 会在 `afterAnalyze` 规划 Hook 中记录 Nereids 已分析的逻辑计划。DML 成功后，系统从该逻辑计划提取血缘并将事件提交到 FE 本地队列。单个 daemon 工作线程会检查每个已加载插件，并将事件分发给 `eventFilter()` 返回 `true` 的插件。 + +![数据血缘采集架构：受支持的 DML 成功后，由 Nereids 分析并提取为 LineageInfo，经 FE 队列和插件投递到外部治理系统。](/images/data-lineage/lineage-architecture-zh-CN.svg) + +在提取前，Doris 会对已加载插件调用 `eventFilter()`。如果没有插件愿意接收事件，则跳过提取。工作线程在分发前也会再次判断 `eventFilter()`。 + +### 血缘内容 + +每个 `LineageInfo` 事件包含表级血缘、列级直接血缘、两类间接血缘和查询上下文。先通过一个完整的 SQL 示例明确源表、目标表和输出结果，后续各小节都基于该示例解释事件内容。 + +#### 示例 SQL 和执行结果 + +以下示例需要当前用户具有创建数据库、创建表、写入和查询权限，并且 FE 已加载血缘插件。示例使用两个源表，通过 CTE、Join、过滤、聚合、窗口函数、条件表达式和排序生成客户汇总表。 + +```sql +CREATE DATABASE IF NOT EXISTS lineage_demo; +USE lineage_demo; + +CREATE TABLE lineage_orders ( + order_id BIGINT, + customer_id BIGINT, + region VARCHAR(16), + amount DECIMAL(18, 2), + status VARCHAR(16) +) +DUPLICATE KEY(order_id) +DISTRIBUTED BY HASH(order_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_customers ( + customer_id BIGINT, + customer_name VARCHAR(64), + customer_level VARCHAR(16) +) +DUPLICATE KEY(customer_id) +DISTRIBUTED BY HASH(customer_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_customer_summary ( + customer_id BIGINT, + region_label VARCHAR(16), + total_amount DECIMAL(18, 2), + customer_seq BIGINT, + region_group VARCHAR(16) +) +DUPLICATE KEY(customer_id) +DISTRIBUTED BY HASH(customer_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +INSERT INTO lineage_customers VALUES + (1, 'Alice', 'VIP'), + (2, 'Bob', 'VIP'), + (3, 'Carol', 'REGULAR'); + +INSERT INTO lineage_orders VALUES + (101, 1, 'east', 100.00, 'PAID'), + (102, 1, 'east', 50.00, 'PAID'), + (103, 2, 'west', 80.00, 'PAID'), + (104, 2, 'west', 30.00, 'CANCELLED'), + (105, 3, 'north', 200.00, 'PAID'); + +INSERT INTO lineage_customer_summary +WITH customer_totals AS ( + SELECT + o.customer_id, + UPPER(o.region) AS region_label, + SUM(o.amount) AS total_amount + FROM lineage_orders o + JOIN lineage_customers c + ON o.customer_id = c.customer_id + WHERE o.status = 'PAID' + AND c.customer_level = 'VIP' + GROUP BY o.customer_id, UPPER(o.region) + HAVING SUM(o.amount) >= 50 +) +SELECT + customer_id, + region_label, + total_amount, + ROW_NUMBER() OVER (ORDER BY customer_id) AS customer_seq, + CASE + WHEN region_label = 'EAST' THEN 'CORE' + ELSE 'OTHER' + END AS region_group +FROM customer_totals +ORDER BY total_amount DESC; + +SELECT customer_id, region_label, total_amount, customer_seq, region_group +FROM lineage_customer_summary +ORDER BY customer_id; +``` + +查询结果如下： + +```text ++-------------+--------------+--------------+--------------+--------------+ +| customer_id | region_label | total_amount | customer_seq | region_group | ++-------------+--------------+--------------+--------------+--------------+ +| 1 | EAST | 150.00 | 1 | CORE | +| 2 | WEST | 80.00 | 2 | OTHER | ++-------------+--------------+--------------+--------------+--------------+ +``` + +两个仅包含 `VALUES` 的 Insert 不会生成血缘事件。`INSERT INTO lineage_customer_summary ... SELECT` 会生成一个 `LineageInfo` 事件，以下各小节说明该事件的具体内容。 + +#### 表级血缘 + +表级血缘记录一个目标表和已分析逻辑计划中扫描的所有源表。例如，写入 `lineage_customer_summary` 的查询同时扫描 `lineage_orders` 和 `lineage_customers`，因此这两个源表都与目标表建立表级关系。CTE 会继续解析到其底层表，`UNION` 的各个分支也都会计入源表集合。 + +该示例产生以下表级血缘： + +| 源表 | 目标表 | +| --- | --- | +| `lineage_orders` | `lineage_customer_summary` | +| `lineage_customers` | `lineage_customer_summary` | + +#### 列级直接血缘 + +列级直接血缘说明每个目标列的值由哪个源表达式产生。Doris 在解析别名和 CTE 后，按以下顺序判断类型：表达式只要包含聚合函数就是 `AGGREGATION`；纯源列引用是 `IDENTITY`；其他表达式是 `TRANSFORMATION`。 + +| 类型 | 含义 | 示例 | +| --- | --- | --- | +| `IDENTITY` | 目标值直接取自一个源列，没有函数或运算。 | `lineage_customer_summary.customer_id` 直接来自 `lineage_orders.customer_id`。 | +| `TRANSFORMATION` | 目标值经过非聚合表达式转换，例如算术、字符串函数、窗口函数或条件表达式。 | `region_label` 来自 `UPPER(lineage_orders.region)`；`customer_seq` 来自 `ROW_NUMBER()`。 | +| `AGGREGATION` | 目标值由包含聚合函数的表达式产生。即使聚合函数外还有其他表达式，仍归为该类型。 | `total_amount` 来自 `SUM(lineage_orders.amount)`。 | + +该示例产生以下列级直接血缘： + +| 目标列 | 源表达式 | 类型 | +| --- | --- | --- | +| `lineage_customer_summary.customer_id` | `lineage_orders.customer_id` | `IDENTITY` | +| `lineage_customer_summary.region_label` | `UPPER(lineage_orders.region)` | `TRANSFORMATION` | +| `lineage_customer_summary.total_amount` | `SUM(lineage_orders.amount)` | `AGGREGATION` | +| `lineage_customer_summary.customer_seq` | `ROW_NUMBER() OVER (ORDER BY lineage_orders.customer_id)` | `TRANSFORMATION` | +| `lineage_customer_summary.region_group` | `CASE WHEN UPPER(lineage_orders.region) = 'EAST' ... END` | `TRANSFORMATION` | + +#### 数据集级间接血缘 + +数据集级间接血缘记录影响整个结果集、但不直接产生某个目标列值的表达式。这些依赖保存在事件级 map 中，需要时可以应用到每个目标输出列。 + +| 类型 | 含义 | 示例 | +| --- | --- | --- | +| `JOIN` | Join 条件决定哪些源表行可以组合。 | `lineage_orders.customer_id = lineage_customers.customer_id`。 | +| `FILTER` | `WHERE` 或 `HAVING` 条件决定哪些行或分组进入结果。 | `status = 'PAID'`、`customer_level = 'VIP'`、`HAVING SUM(amount) >= 50`。 | +| `GROUP_BY` | 分组表达式决定聚合结果的粒度。 | 按 `customer_id` 和 `UPPER(region)` 分组。 | +| `SORT` | `ORDER BY` 或 TopN 的排序表达式决定结果顺序。 | 按 `total_amount DESC` 排序；别名解析后对应 `SUM(amount)`。 | + +该示例产生以下数据集级间接血缘： + +| 类型 | 事件中的源表达式示例 | 作用 | +| --- | --- | --- | +| `JOIN` | `lineage_orders.customer_id = lineage_customers.customer_id` | 决定订单与客户的匹配关系。 | +| `FILTER` | `status = 'PAID'`、`customer_level = 'VIP'`、`SUM(amount) >= 50` | 分别来自 `WHERE` 和 `HAVING`，共同决定进入结果的行和分组。 | +| `GROUP_BY` | `lineage_orders.customer_id`、`UPPER(lineage_orders.region)` | 决定每个客户、区域的聚合粒度。 | +| `SORT` | `total_amount DESC`，解析后对应 `SUM(lineage_orders.amount)` | 决定插入查询结果的顺序。 | + +#### 输出列级间接血缘 + +输出列级间接血缘只挂到受影响的目标列，不会应用到其他输出列。 + +| 类型 | 含义 | 示例 | +| --- | --- | --- | +| `WINDOW` | 窗口函数的 `PARTITION BY` 和 `ORDER BY` 输入影响对应输出列。 | `customer_seq` 由 `ROW_NUMBER() OVER (ORDER BY customer_id)` 产生，其 `WINDOW` 依赖是 `customer_id`。 | +| `CONDITIONAL` | `CASE`、`IF` 的条件或 `COALESCE` 的候选表达式影响对应输出列。 | `region_group` 的值由 `CASE WHEN region_label = 'EAST' ...` 决定，其 `CONDITIONAL` 依赖是该判断条件。 | + +窗口或条件表达式本身仍会出现在列级直接血缘中。例如，`customer_seq` 的直接类型是 `TRANSFORMATION`，同时具有 `WINDOW` 间接血缘；`region_group` 的直接类型也是 `TRANSFORMATION`，同时具有 `CONDITIONAL` 间接血缘。 + +该示例产生以下输出列级间接血缘： + +| 目标列 | 类型 | 事件中的源表达式 | 作用 | +| --- | --- | --- | --- | +| `customer_seq` | `WINDOW` | `lineage_orders.customer_id` | 作为 `ROW_NUMBER()` 的窗口排序输入，只影响 `customer_seq`。 | +| `region_group` | `CONDITIONAL` | `UPPER(lineage_orders.region) = 'EAST'` | 作为 `CASE WHEN` 条件，只影响 `region_group`。 | + +其他三个目标列没有输出列级间接血缘。数据集级的 `JOIN`、`FILTER`、`GROUP_BY` 和 `SORT` 仍然影响整个结果集。 + +#### 查询上下文 + +查询上下文说明这次血缘事件由谁、在什么会话中、通过哪条 DML 产生。 + +| 字段 | 示例或含义 | +| --- | --- | +| Source Command | `InsertIntoTableCommand`，表示事件来自 Insert。 | +| Query ID 和原始 SQL | 本次 DML 的 Query ID，以及完整的 `INSERT INTO lineage_customer_summary ...` 文本。 | +| 用户和客户端 IP | 例如 `lineage_user` 和 `192.0.2.10`。 | +| 会话数据库和 Catalog | 上述示例为 `lineage_demo` 和 `internal`；它们是会话上下文，不一定等于目标表所在位置。 | +| 执行状态 | 成功的 DML 通常记录为 `OK`。 | +| 时间戳和耗时 | 事件创建时间和从查询开始到事件创建的毫秒数；无法取得时为 `-1`。 | +| 外部 Catalog 属性 | 记录查询涉及的外部 Catalog 的非敏感属性。密码、密钥和隐藏属性会被删除；仅使用 `internal` Catalog 时该 map 为空。 | + +该示例的原始 SQL 是上述复杂 Insert，Source Command 为 `InsertIntoTableCommand`，会话数据库为 `lineage_demo`，会话 Catalog 为 `internal`，成功状态为 `OK`。Query ID、用户、客户端 IP、事件时间戳和耗时使用实际执行上下文中的值；因为示例只使用 Internal Catalog，外部 Catalog 属性 map 为空。 + +在插件的下游系统中确认已收到事件。首次部署时，还应检查 `fe.log` 是否出现 `Loaded lineage plugin`，以及插件自身错误或队列满 warning。 + +![LineageInfo 事件模型：目标和源表形成直接、间接血缘，并与查询上下文一起由插件转换为下游系统事件。](/images/data-lineage/lineage-event-model-zh-CN.svg) + +提取器会在事件到达插件前解析 CTE 生产者表达式，并展开 `UNION` 的各分支。下游插件需要将 `TableIf`、`SlotReference` 和 `Expression` 等 Doris Java 对象转换为稳定标识或自身事件格式。 + +### 事件处理行为 + +`lineage_event_queue_size` 设置每个 FE 进程本地队列能够等待处理的最大血缘事件数，单位是事件数，不是字节数。队列满时，系统会丢弃新事件，DML 正常继续。工作线程串行调用插件，慢插件会延迟后续事件，并可能在持续负载下造成事件丢弃。插件抛出的异常会被记录，不会停止工作线程，也不会影响 DML。 + +插件发现和初始化只在 FE 启动时进行，不支持动态 reload 或 unload。 + +## 配置和使用 + +### 前提条件 + +1. 准备外部血缘插件。Doris 不提供内置 sink。SPI 契约、JAR 打包方式和完整最小插件示例请参阅[数据血缘插件开发](/community/developer-guide/data-lineage-plugin-development)。 +2. 将插件 JAR 及所需第三方依赖 JAR 复制到每个 FE。 +3. 如果插件需要向外部治理服务发送事件，确保每个可能执行 DML 的 FE 都可以访问该服务端点。 + +### 部署插件 + +`FE_HOME` 表示包含 `bin/`、`conf/` 和 `lib/` 的单个 FE 安装目录。在每个 FE 上使用以下目录结构。加载器只扫描 `lineage/` 下的直接子目录，以及插件目录和其 `lib/` 目录中的 JAR。 + +```text +$FE_HOME/plugins/ +└── lineage/ + └── example-lineage/ + ├── example-lineage.jar + └── lib/ + └── downstream-client.jar +``` + +在每个 FE 的 `fe.conf` 中配置。将 `example-lineage` 替换为插件 Factory 的 `name()` 返回值： + +```text +plugin_dir = /opt/apache-doris/fe/plugins +activate_lineage_plugin = example-lineage +lineage_event_queue_size = 50000 +``` + +| 配置 | 类型和默认值 | 是否必填 | 说明 | +| --- | --- | --- | --- | +| `plugin_dir` | String；`$FE_HOME/plugins` | 否 | 插件根目录。血缘加载器扫描其 `lineage/` 直接子目录。使用默认目录时可以省略该配置。 | +| `activate_lineage_plugin` | String 数组；空 | 建议显式填写 | 需要实例化的 Factory 名称，以英文逗号分隔。空值表示不过滤，会实例化全部已发现的 Factory。 | +| `lineage_event_queue_size` | 正整数；`50000` | 否 | 当前 FE 等待工作线程处理的最大血缘事件数。队列满时丢弃新事件。 | + +三个参数都在各 FE 节点的 `$FE_HOME/conf/fe.conf` 中配置，作用范围是当前 FE 进程。它们都只在 FE 启动时生效，修改后必须重启对应 FE。 + +#### 激活插件 + +在每个 FE 节点的 `$FE_HOME/conf/fe.conf` 中配置 `activate_lineage_plugin`。它控制哪些已发现的 Factory 会被实例化，不负责发现 JAR，也不控制队列容量。名称必须与 `LineagePluginFactory.name()` 完全一致并区分大小写；插件实现的 `LineagePlugin.name()` 应返回相同名称。插件目录名不参与匹配。配置多个插件时使用英文逗号分隔；FE 配置解析器会去除各项两侧的空格，但名称匹配仍区分大小写： + +```text +activate_lineage_plugin = example-lineage,governance-lineage +``` + +FE 启动时会发现内置和外部 Factory，然后只为配置中列出的名称创建插件实例。每个 FE 独立读取该配置，因此所有可能执行 DML 的 FE 都需要配置相同的插件集合。修改后必须重启 FE，不支持通过 `ADMIN SET FRONTEND CONFIG` 动态修改。 + +:::caution 空值行为 + +当前实现中，`activate_lineage_plugin` 为空并不表示关闭血缘插件，而是跳过名称筛选并实例化全部已发现的 Factory。应显式填写需要启用的插件名称，不要把空值当作禁用开关。插件实例加载后，`eventFilter()` 仍会在提取前和分发前决定是否接收事件。 + +::: + +#### 配置事件队列 + +在每个 FE 节点的 `$FE_HOME/conf/fe.conf` 中配置 `lineage_event_queue_size`。该参数必须是正整数，每个可能执行 DML 的 FE 都有自己的独立队列，因此需要分别配置。 + +该参数是 FE 启动配置，不支持通过 `ADMIN SET FRONTEND CONFIG` 动态修改。修改后必须重启对应 FE。增大参数只会提高待处理事件的缓冲能力和 FE 内存占用，不会增加消费线程数，也不会提高插件处理速度。出现事件丢失时，应先在 `fe.log` 中搜索 `the lineage event queue is full`，并优先降低插件 `exec()` 的处理延迟，再根据峰值积压量和 FE 内存容量调整队列大小。 + +修改其他插件配置或替换插件 JAR 后，也必须重启每个 FE。这些配置和插件目录只在 FE 启动时读取。 + +## 运维和排障 + +### 写入后没有事件 + +常见原因包括语句不受支持、DML 执行失败、当前 FE 没有加载插件，或者插件的 `eventFilter()` 返回了 `false`。 + +按以下顺序检查： + +1. 确认语句是成功执行的 `INSERT INTO ... SELECT`、`INSERT OVERWRITE TABLE ... SELECT` 或 `CREATE TABLE AS SELECT`。仅包含 `VALUES` 的 Insert 不会生成事件。 +2. 确认执行该 DML 的 FE 已配置并加载插件，在 `fe.log` 中搜索 `Loaded lineage plugin`。 +3. 确认插件的 `eventFilter()` 在查询线程和工作线程中都返回 `true`。 +4. 检查插件日志和下游服务，确认事件不是在 FE 之外处理失败。 + +### 插件未加载 + +检查插件目录中是否存在 JAR，以及 JAR 中是否包含 `META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory`。每个插件目录只能提供一个 Factory，Factory 名称必须全局唯一，并且应与 `activate_lineage_plugin` 中的名称一致。修正目录、JAR 或配置后重启对应 FE。 + +### 高负载下事件缺失 + +在 `fe.log` 中搜索 `the lineage event queue is full` 和插件异常。如果队列已满，先减少插件 `exec()` 中的同步阻塞，使下游投递具备幂等性和有界重试，再评估是否增大 `lineage_event_queue_size`。增大队列前需要评估 FE 内存，因为不同 SQL 产生的事件大小并不固定。 + +还应确认所有可能执行 DML 的 FE 都部署了相同插件和配置。每个 FE 使用独立的本地队列，一个 FE 上的插件和队列状态不会自动同步到其他 FE。 + +### 更新插件后未生效 + +插件目录只在 FE 启动时扫描，不支持动态 reload 或 unload。替换插件 JAR、依赖或配置后，重启每个相关 FE，并再次检查 `Loaded lineage plugin` 日志。 + +应使用 Query ID 和目标表等信息构造下游事件标识。这样，即使插件在框架外实现重试，治理系统也可以对事件去重。 + +## 相关文档 + +- [数据血缘插件开发](/community/developer-guide/data-lineage-plugin-development) diff --git a/i18n/zh-CN/docusaurus-plugin-content-docs/version-4.x/data-governance/data-lineage.md b/i18n/zh-CN/docusaurus-plugin-content-docs/version-4.x/data-governance/data-lineage.md new file mode 100644 index 0000000000000..931577a9a057d --- /dev/null +++ b/i18n/zh-CN/docusaurus-plugin-content-docs/version-4.x/data-governance/data-lineage.md @@ -0,0 +1,336 @@ +--- +{ + "title": "数据血缘原理与使用（4.x）", + "sidebar_label": "数据血缘", + "language": "zh-CN", + "description": "介绍 Apache Doris 数据血缘的采集原理、事件内容、支持范围、插件部署、FE 配置、队列行为、运维排障和完整 SQL 示例，帮助用户理解并验证表级、列级以及直接和间接血缘关系。", + "keywords": ["Apache Doris", "数据血缘", "数据治理", "LineagePlugin", "列级血缘"] +} +--- + + + + +数据血缘功能可以从 Doris 支持的 DML 语句中提取表级和列级依赖，并通过 `LineagePlugin` 发送到外部治理系统。Doris 是血缘生产者，不保存事件、不提供血缘查询 API，也不提供血缘可视化。 + +> 该功能从 Apache Doris 4.0.6 起在 4.x 版本中可用。 + +## 能力和限制 + +框架只会在以下语句成功后生成事件： + +| 支持的语句 | 事件行为 | +| --- | --- | +| `INSERT INTO ... SELECT` | Insert 成功后生成一个事件。 | +| `INSERT OVERWRITE TABLE ... SELECT` | Overwrite 成功后生成一个事件。 | +| `CREATE TABLE AS SELECT` | 内部 Insert 成功后生成一个事件。 | + +当前实现不会为 `SELECT`、`UPDATE`、`DELETE`、导入任务、仅包含 `VALUES` 的 Insert，以及目标表为 `__internal_schema` 的写入生成事件。部分 `UPDATE` 和 `DELETE` 执行路径会在内部复用 Insert 命令，但原始命令类型校验会阻止这些命令提交血缘事件。 + +:::caution 投递保证 + +血缘投递是异步且尽力而为的。DML 成功后，即使事件采集或插件投递失败，DML 也不会回滚。队列满时可能丢弃事件，Doris 不会重试或持久化这些事件。 + +::: + +## 工作原理 + +### 采集流程 + +对于受支持的语句，Doris 会在 `afterAnalyze` 规划 Hook 中记录 Nereids 已分析的逻辑计划。DML 成功后，系统从该逻辑计划提取血缘并将事件提交到 FE 本地队列。单个 daemon 工作线程会检查每个已加载插件，并将事件分发给 `eventFilter()` 返回 `true` 的插件。 + +![数据血缘采集架构：受支持的 DML 成功后，由 Nereids 分析并提取为 LineageInfo，经 FE 队列和插件投递到外部治理系统。](/images/data-lineage/lineage-architecture-zh-CN.svg) + +在提取前，Doris 会对已加载插件调用 `eventFilter()`。如果没有插件愿意接收事件，则跳过提取。工作线程在分发前也会再次判断 `eventFilter()`。 + +### 血缘内容 + +每个 `LineageInfo` 事件包含表级血缘、列级直接血缘、两类间接血缘和查询上下文。先通过一个完整的 SQL 示例明确源表、目标表和输出结果，后续各小节都基于该示例解释事件内容。 + +#### 示例 SQL 和执行结果 + +以下示例需要当前用户具有创建数据库、创建表、写入和查询权限，并且 FE 已加载血缘插件。示例使用两个源表，通过 CTE、Join、过滤、聚合、窗口函数、条件表达式和排序生成客户汇总表。 + +```sql +CREATE DATABASE IF NOT EXISTS lineage_demo; +USE lineage_demo; + +CREATE TABLE lineage_orders ( + order_id BIGINT, + customer_id BIGINT, + region VARCHAR(16), + amount DECIMAL(18, 2), + status VARCHAR(16) +) +DUPLICATE KEY(order_id) +DISTRIBUTED BY HASH(order_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_customers ( + customer_id BIGINT, + customer_name VARCHAR(64), + customer_level VARCHAR(16) +) +DUPLICATE KEY(customer_id) +DISTRIBUTED BY HASH(customer_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_customer_summary ( + customer_id BIGINT, + region_label VARCHAR(16), + total_amount DECIMAL(18, 2), + customer_seq BIGINT, + region_group VARCHAR(16) +) +DUPLICATE KEY(customer_id) +DISTRIBUTED BY HASH(customer_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +INSERT INTO lineage_customers VALUES + (1, 'Alice', 'VIP'), + (2, 'Bob', 'VIP'), + (3, 'Carol', 'REGULAR'); + +INSERT INTO lineage_orders VALUES + (101, 1, 'east', 100.00, 'PAID'), + (102, 1, 'east', 50.00, 'PAID'), + (103, 2, 'west', 80.00, 'PAID'), + (104, 2, 'west', 30.00, 'CANCELLED'), + (105, 3, 'north', 200.00, 'PAID'); + +INSERT INTO lineage_customer_summary +WITH customer_totals AS ( + SELECT + o.customer_id, + UPPER(o.region) AS region_label, + SUM(o.amount) AS total_amount + FROM lineage_orders o + JOIN lineage_customers c + ON o.customer_id = c.customer_id + WHERE o.status = 'PAID' + AND c.customer_level = 'VIP' + GROUP BY o.customer_id, UPPER(o.region) + HAVING SUM(o.amount) >= 50 +) +SELECT + customer_id, + region_label, + total_amount, + ROW_NUMBER() OVER (ORDER BY customer_id) AS customer_seq, + CASE + WHEN region_label = 'EAST' THEN 'CORE' + ELSE 'OTHER' + END AS region_group +FROM customer_totals +ORDER BY total_amount DESC; + +SELECT customer_id, region_label, total_amount, customer_seq, region_group +FROM lineage_customer_summary +ORDER BY customer_id; +``` + +查询结果如下： + +```text ++-------------+--------------+--------------+--------------+--------------+ +| customer_id | region_label | total_amount | customer_seq | region_group | ++-------------+--------------+--------------+--------------+--------------+ +| 1 | EAST | 150.00 | 1 | CORE | +| 2 | WEST | 80.00 | 2 | OTHER | ++-------------+--------------+--------------+--------------+--------------+ +``` + +两个仅包含 `VALUES` 的 Insert 不会生成血缘事件。`INSERT INTO lineage_customer_summary ... SELECT` 会生成一个 `LineageInfo` 事件，以下各小节说明该事件的具体内容。 + +#### 表级血缘 + +表级血缘记录一个目标表和已分析逻辑计划中扫描的所有源表。例如，写入 `lineage_customer_summary` 的查询同时扫描 `lineage_orders` 和 `lineage_customers`，因此这两个源表都与目标表建立表级关系。CTE 会继续解析到其底层表，`UNION` 的各个分支也都会计入源表集合。 + +该示例产生以下表级血缘： + +| 源表 | 目标表 | +| --- | --- | +| `lineage_orders` | `lineage_customer_summary` | +| `lineage_customers` | `lineage_customer_summary` | + +#### 列级直接血缘 + +列级直接血缘说明每个目标列的值由哪个源表达式产生。Doris 在解析别名和 CTE 后，按以下顺序判断类型：表达式只要包含聚合函数就是 `AGGREGATION`；纯源列引用是 `IDENTITY`；其他表达式是 `TRANSFORMATION`。 + +| 类型 | 含义 | 示例 | +| --- | --- | --- | +| `IDENTITY` | 目标值直接取自一个源列，没有函数或运算。 | `lineage_customer_summary.customer_id` 直接来自 `lineage_orders.customer_id`。 | +| `TRANSFORMATION` | 目标值经过非聚合表达式转换，例如算术、字符串函数、窗口函数或条件表达式。 | `region_label` 来自 `UPPER(lineage_orders.region)`；`customer_seq` 来自 `ROW_NUMBER()`。 | +| `AGGREGATION` | 目标值由包含聚合函数的表达式产生。即使聚合函数外还有其他表达式，仍归为该类型。 | `total_amount` 来自 `SUM(lineage_orders.amount)`。 | + +该示例产生以下列级直接血缘： + +| 目标列 | 源表达式 | 类型 | +| --- | --- | --- | +| `lineage_customer_summary.customer_id` | `lineage_orders.customer_id` | `IDENTITY` | +| `lineage_customer_summary.region_label` | `UPPER(lineage_orders.region)` | `TRANSFORMATION` | +| `lineage_customer_summary.total_amount` | `SUM(lineage_orders.amount)` | `AGGREGATION` | +| `lineage_customer_summary.customer_seq` | `ROW_NUMBER() OVER (ORDER BY lineage_orders.customer_id)` | `TRANSFORMATION` | +| `lineage_customer_summary.region_group` | `CASE WHEN UPPER(lineage_orders.region) = 'EAST' ... END` | `TRANSFORMATION` | + +#### 数据集级间接血缘 + +数据集级间接血缘记录影响整个结果集、但不直接产生某个目标列值的表达式。这些依赖保存在事件级 map 中，需要时可以应用到每个目标输出列。 + +| 类型 | 含义 | 示例 | +| --- | --- | --- | +| `JOIN` | Join 条件决定哪些源表行可以组合。 | `lineage_orders.customer_id = lineage_customers.customer_id`。 | +| `FILTER` | `WHERE` 或 `HAVING` 条件决定哪些行或分组进入结果。 | `status = 'PAID'`、`customer_level = 'VIP'`、`HAVING SUM(amount) >= 50`。 | +| `GROUP_BY` | 分组表达式决定聚合结果的粒度。 | 按 `customer_id` 和 `UPPER(region)` 分组。 | +| `SORT` | `ORDER BY` 或 TopN 的排序表达式决定结果顺序。 | 按 `total_amount DESC` 排序；别名解析后对应 `SUM(amount)`。 | + +该示例产生以下数据集级间接血缘： + +| 类型 | 事件中的源表达式示例 | 作用 | +| --- | --- | --- | +| `JOIN` | `lineage_orders.customer_id = lineage_customers.customer_id` | 决定订单与客户的匹配关系。 | +| `FILTER` | `status = 'PAID'`、`customer_level = 'VIP'`、`SUM(amount) >= 50` | 分别来自 `WHERE` 和 `HAVING`，共同决定进入结果的行和分组。 | +| `GROUP_BY` | `lineage_orders.customer_id`、`UPPER(lineage_orders.region)` | 决定每个客户、区域的聚合粒度。 | +| `SORT` | `total_amount DESC`，解析后对应 `SUM(lineage_orders.amount)` | 决定插入查询结果的顺序。 | + +#### 输出列级间接血缘 + +输出列级间接血缘只挂到受影响的目标列，不会应用到其他输出列。 + +| 类型 | 含义 | 示例 | +| --- | --- | --- | +| `WINDOW` | 窗口函数的 `PARTITION BY` 和 `ORDER BY` 输入影响对应输出列。 | `customer_seq` 由 `ROW_NUMBER() OVER (ORDER BY customer_id)` 产生，其 `WINDOW` 依赖是 `customer_id`。 | +| `CONDITIONAL` | `CASE`、`IF` 的条件或 `COALESCE` 的候选表达式影响对应输出列。 | `region_group` 的值由 `CASE WHEN region_label = 'EAST' ...` 决定，其 `CONDITIONAL` 依赖是该判断条件。 | + +窗口或条件表达式本身仍会出现在列级直接血缘中。例如，`customer_seq` 的直接类型是 `TRANSFORMATION`，同时具有 `WINDOW` 间接血缘；`region_group` 的直接类型也是 `TRANSFORMATION`，同时具有 `CONDITIONAL` 间接血缘。 + +该示例产生以下输出列级间接血缘： + +| 目标列 | 类型 | 事件中的源表达式 | 作用 | +| --- | --- | --- | --- | +| `customer_seq` | `WINDOW` | `lineage_orders.customer_id` | 作为 `ROW_NUMBER()` 的窗口排序输入，只影响 `customer_seq`。 | +| `region_group` | `CONDITIONAL` | `UPPER(lineage_orders.region) = 'EAST'` | 作为 `CASE WHEN` 条件，只影响 `region_group`。 | + +其他三个目标列没有输出列级间接血缘。数据集级的 `JOIN`、`FILTER`、`GROUP_BY` 和 `SORT` 仍然影响整个结果集。 + +#### 查询上下文 + +查询上下文说明这次血缘事件由谁、在什么会话中、通过哪条 DML 产生。 + +| 字段 | 示例或含义 | +| --- | --- | +| Source Command | `InsertIntoTableCommand`，表示事件来自 Insert。 | +| Query ID 和原始 SQL | 本次 DML 的 Query ID，以及完整的 `INSERT INTO lineage_customer_summary ...` 文本。 | +| 用户和客户端 IP | 例如 `lineage_user` 和 `192.0.2.10`。 | +| 会话数据库和 Catalog | 上述示例为 `lineage_demo` 和 `internal`；它们是会话上下文，不一定等于目标表所在位置。 | +| 执行状态 | 成功的 DML 通常记录为 `OK`。 | +| 时间戳和耗时 | 事件创建时间和从查询开始到事件创建的毫秒数；无法取得时为 `-1`。 | +| 外部 Catalog 属性 | 记录查询涉及的外部 Catalog 的非敏感属性。密码、密钥和隐藏属性会被删除；仅使用 `internal` Catalog 时该 map 为空。 | + +该示例的原始 SQL 是上述复杂 Insert，Source Command 为 `InsertIntoTableCommand`，会话数据库为 `lineage_demo`，会话 Catalog 为 `internal`，成功状态为 `OK`。Query ID、用户、客户端 IP、事件时间戳和耗时使用实际执行上下文中的值；因为示例只使用 Internal Catalog，外部 Catalog 属性 map 为空。 + +在插件的下游系统中确认已收到事件。首次部署时，还应检查 `fe.log` 是否出现 `Loaded lineage plugin`，以及插件自身错误或队列满 warning。 + +![LineageInfo 事件模型：目标和源表形成直接、间接血缘，并与查询上下文一起由插件转换为下游系统事件。](/images/data-lineage/lineage-event-model-zh-CN.svg) + +提取器会在事件到达插件前解析 CTE 生产者表达式，并展开 `UNION` 的各分支。下游插件需要将 `TableIf`、`SlotReference` 和 `Expression` 等 Doris Java 对象转换为稳定标识或自身事件格式。 + +### 事件处理行为 + +`lineage_event_queue_size` 设置每个 FE 进程本地队列能够等待处理的最大血缘事件数，单位是事件数，不是字节数。队列满时，系统会丢弃新事件，DML 正常继续。工作线程串行调用插件，慢插件会延迟后续事件，并可能在持续负载下造成事件丢弃。插件抛出的异常会被记录，不会停止工作线程，也不会影响 DML。 + +插件发现和初始化只在 FE 启动时进行，不支持动态 reload 或 unload。 + +## 配置和使用 + +### 前提条件 + +1. 准备外部血缘插件。Doris 不提供内置 sink。SPI 契约、JAR 打包方式和完整最小插件示例请参阅[数据血缘插件开发](/community/developer-guide/data-lineage-plugin-development)。 +2. 将插件 JAR 及所需第三方依赖 JAR 复制到每个 FE。 +3. 如果插件需要向外部治理服务发送事件，确保每个可能执行 DML 的 FE 都可以访问该服务端点。 + +### 部署插件 + +`FE_HOME` 表示包含 `bin/`、`conf/` 和 `lib/` 的单个 FE 安装目录。在每个 FE 上使用以下目录结构。加载器只扫描 `lineage/` 下的直接子目录，以及插件目录和其 `lib/` 目录中的 JAR。 + +```text +$FE_HOME/plugins/ +└── lineage/ + └── example-lineage/ + ├── example-lineage.jar + └── lib/ + └── downstream-client.jar +``` + +在每个 FE 的 `fe.conf` 中配置。将 `example-lineage` 替换为插件 Factory 的 `name()` 返回值： + +```text +plugin_dir = /opt/apache-doris/fe/plugins +activate_lineage_plugin = example-lineage +lineage_event_queue_size = 50000 +``` + +| 配置 | 类型和默认值 | 是否必填 | 说明 | +| --- | --- | --- | --- | +| `plugin_dir` | String；`$FE_HOME/plugins` | 否 | 插件根目录。血缘加载器扫描其 `lineage/` 直接子目录。使用默认目录时可以省略该配置。 | +| `activate_lineage_plugin` | String 数组；空 | 建议显式填写 | 需要实例化的 Factory 名称，以英文逗号分隔。空值表示不过滤，会实例化全部已发现的 Factory。 | +| `lineage_event_queue_size` | 正整数；`50000` | 否 | 当前 FE 等待工作线程处理的最大血缘事件数。队列满时丢弃新事件。 | + +三个参数都在各 FE 节点的 `$FE_HOME/conf/fe.conf` 中配置，作用范围是当前 FE 进程。它们都只在 FE 启动时生效，修改后必须重启对应 FE。 + +#### 激活插件 + +在每个 FE 节点的 `$FE_HOME/conf/fe.conf` 中配置 `activate_lineage_plugin`。它控制哪些已发现的 Factory 会被实例化，不负责发现 JAR，也不控制队列容量。名称必须与 `LineagePluginFactory.name()` 完全一致并区分大小写；插件实现的 `LineagePlugin.name()` 应返回相同名称。插件目录名不参与匹配。配置多个插件时使用英文逗号分隔；FE 配置解析器会去除各项两侧的空格，但名称匹配仍区分大小写： + +```text +activate_lineage_plugin = example-lineage,governance-lineage +``` + +FE 启动时会发现内置和外部 Factory，然后只为配置中列出的名称创建插件实例。每个 FE 独立读取该配置，因此所有可能执行 DML 的 FE 都需要配置相同的插件集合。修改后必须重启 FE，不支持通过 `ADMIN SET FRONTEND CONFIG` 动态修改。 + +:::caution 空值行为 + +当前实现中，`activate_lineage_plugin` 为空并不表示关闭血缘插件，而是跳过名称筛选并实例化全部已发现的 Factory。应显式填写需要启用的插件名称，不要把空值当作禁用开关。插件实例加载后，`eventFilter()` 仍会在提取前和分发前决定是否接收事件。 + +::: + +#### 配置事件队列 + +在每个 FE 节点的 `$FE_HOME/conf/fe.conf` 中配置 `lineage_event_queue_size`。该参数必须是正整数，每个可能执行 DML 的 FE 都有自己的独立队列，因此需要分别配置。 + +该参数是 FE 启动配置，不支持通过 `ADMIN SET FRONTEND CONFIG` 动态修改。修改后必须重启对应 FE。增大参数只会提高待处理事件的缓冲能力和 FE 内存占用，不会增加消费线程数，也不会提高插件处理速度。出现事件丢失时，应先在 `fe.log` 中搜索 `the lineage event queue is full`，并优先降低插件 `exec()` 的处理延迟，再根据峰值积压量和 FE 内存容量调整队列大小。 + +修改其他插件配置或替换插件 JAR 后，也必须重启每个 FE。这些配置和插件目录只在 FE 启动时读取。 + +## 运维和排障 + +### 写入后没有事件 + +常见原因包括语句不受支持、DML 执行失败、当前 FE 没有加载插件，或者插件的 `eventFilter()` 返回了 `false`。 + +按以下顺序检查： + +1. 确认语句是成功执行的 `INSERT INTO ... SELECT`、`INSERT OVERWRITE TABLE ... SELECT` 或 `CREATE TABLE AS SELECT`。仅包含 `VALUES` 的 Insert 不会生成事件。 +2. 确认执行该 DML 的 FE 已配置并加载插件，在 `fe.log` 中搜索 `Loaded lineage plugin`。 +3. 确认插件的 `eventFilter()` 在查询线程和工作线程中都返回 `true`。 +4. 检查插件日志和下游服务，确认事件不是在 FE 之外处理失败。 + +### 插件未加载 + +检查插件目录中是否存在 JAR，以及 JAR 中是否包含 `META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory`。每个插件目录只能提供一个 Factory，Factory 名称必须全局唯一，并且应与 `activate_lineage_plugin` 中的名称一致。修正目录、JAR 或配置后重启对应 FE。 + +### 高负载下事件缺失 + +在 `fe.log` 中搜索 `the lineage event queue is full` 和插件异常。如果队列已满，先减少插件 `exec()` 中的同步阻塞，使下游投递具备幂等性和有界重试，再评估是否增大 `lineage_event_queue_size`。增大队列前需要评估 FE 内存，因为不同 SQL 产生的事件大小并不固定。 + +还应确认所有可能执行 DML 的 FE 都部署了相同插件和配置。每个 FE 使用独立的本地队列，一个 FE 上的插件和队列状态不会自动同步到其他 FE。 + +### 更新插件后未生效 + +插件目录只在 FE 启动时扫描，不支持动态 reload 或 unload。替换插件 JAR、依赖或配置后，重启每个相关 FE，并再次检查 `Loaded lineage plugin` 日志。 + +应使用 Query ID 和目标表等信息构造下游事件标识。这样，即使插件在框架外实现重试，治理系统也可以对事件去重。 + +## 相关文档 + +- [数据血缘插件开发](/community/developer-guide/data-lineage-plugin-development) diff --git a/sidebars.ts b/sidebars.ts index 0a53243f2dd9a..a0d897dd83da6 100644 --- a/sidebars.ts +++ b/sidebars.ts @@ -757,6 +757,13 @@ const sidebars: SidebarsConfig = { ], }, ], + }, + { + type: 'category', + label: 'Data Governance', + items: [ + 'data-governance/data-lineage', + ], }, ], }, diff --git a/sidebarsCommunity.json b/sidebarsCommunity.json index 7ae9daf8ac08e..67cfc62837ca3 100644 --- a/sidebarsCommunity.json +++ b/sidebarsCommunity.json @@ -93,6 +93,13 @@ "developer-guide/data-source-extension/trino-connector-developer-guide" ] }, + { + "type": "category", + "label": "Plugin Development", + "items": [ + "developer-guide/data-lineage-plugin-development" + ] + }, { "type": "category", "label": "Data Format Reference", diff --git a/static/images/data-lineage/lineage-architecture-zh-CN.svg b/static/images/data-lineage/lineage-architecture-zh-CN.svg new file mode 100644 index 0000000000000..ce603c72c1713 --- /dev/null +++ b/static/images/data-lineage/lineage-architecture-zh-CN.svg @@ -0,0 +1,76 @@ + + Apache Doris 数据血缘采集架构 + 受支持的 DML 依次经过 Nereids 分析和执行。仅执行成功后提取 LineageInfo，事件进入 FE 本地有界队列，由单个工作线程交给血缘插件，再发送到外部治理系统。 + + + + + + + + + 数据血缘采集与投递 + 每个 FE 独立采集和排队；只有 DML 执行成功才生成血缘事件。 + + + 受支持的 DML + INSERT INTO SELECT + INSERT OVERWRITE TABLE + CREATE TABLE AS SELECT + + + Nereids 分析 + 分析逻辑计划 + 规划 Hook 保存计划 + + + DML 执行 + 执行数据写入 + 失败时不生成事件 + + + 提取器 + 生成 LineageInfo + 表、列和查询上下文 + + + FE 本地队列 + 有界事件缓冲区 + 满时丢弃新事件 + + + 工作线程 + 串行分发事件 + 单个 daemon 线程 + + + 血缘插件 + LineagePlugin + eventFilter() / exec() + + + + + 仅成功后继续 + + + + + + 结束，不生成血缘事件 + + 执行失败 + + + 外部数据治理系统 + + diff --git a/static/images/data-lineage/lineage-architecture.svg b/static/images/data-lineage/lineage-architecture.svg new file mode 100644 index 0000000000000..9eb4416e66d96 --- /dev/null +++ b/static/images/data-lineage/lineage-architecture.svg @@ -0,0 +1,65 @@ + + Apache Doris data lineage collection architecture + Supported DML is analyzed by Nereids and, after successful execution, produces a LineageInfo event. The event enters a bounded FE queue, is processed by one worker, and is sent by a lineage plugin to an external governance system. + + + + + + Data Lineage Collection and Delivery + Collection occurs after a supported DML statement succeeds. Delivery is asynchronous and best effort. + + + Supported DML + INSERT INTO SELECT + INSERT OVERWRITE TABLE + CREATE TABLE AS SELECT + + + Nereids analysis + Analyzed logical plan + Stored by the planner hook + + + DML execution + Only successful writes continue + + + Extractor + LineageInfo + Tables, columns, context + + + FE queue + Bounded capacity + Full queue drops new events + + + Worker + One daemon thread + + + Plugin + LineagePlugin + exec() + + + + + + analyzed plan + success only + + + + + Governance system + + diff --git a/static/images/data-lineage/lineage-event-model-zh-CN.svg b/static/images/data-lineage/lineage-event-model-zh-CN.svg new file mode 100644 index 0000000000000..86fb52dd11212 --- /dev/null +++ b/static/images/data-lineage/lineage-event-model-zh-CN.svg @@ -0,0 +1,66 @@ + + LineageInfo 事件模型 + LineageInfo 由表信息、列级直接血缘、数据集级间接血缘、输出列级间接血缘和查询上下文组成。插件将这些 Doris 内存对象转换为稳定的下游事件格式。 + + + + + + + + + LineageInfo 事件模型 + 事件聚合五类血缘和查询信息，再由插件转换为外部治理系统使用的稳定格式。 + + + LineageInfo + Doris FE 内存事件 + + + 表信息 + 目标表和目标输出列 + 查询引用的源表 + 已解析 CTE 和 UNION + + + 列级直接血缘 + IDENTITY：直接源列 + TRANSFORMATION：表达式 + AGGREGATION：聚合 + + + 数据集级间接血缘 + JOIN、FILTER + GROUP_BY、SORT + 影响整个结果集 + + + 输出列级间接血缘 + WINDOW：窗口分区和排序表达式 + CONDITIONAL：条件表达式 + 只影响对应目标输出列 + + + 查询上下文 + Query ID、原始 SQL、用户、客户端 IP + 会话数据库和 Catalog、执行状态、时间和耗时 + 外部 Catalog 属性在进入事件前脱敏 + + + 插件输出 + 稳定数据集 ID + 表级依赖关系 + 列级依赖关系 + 查询上下文 + 下游事件格式 + + + 插件转换 + diff --git a/static/images/data-lineage/lineage-event-model.svg b/static/images/data-lineage/lineage-event-model.svg new file mode 100644 index 0000000000000..0815b386aee2e --- /dev/null +++ b/static/images/data-lineage/lineage-event-model.svg @@ -0,0 +1,64 @@ + + LineageInfo event model + A LineageInfo event combines target and source tables, direct lineage per output column, indirect lineage, and query context. + + + + + + LineageInfo Event Model + A plugin converts these in-memory Doris objects into a stable downstream event schema. + + + Target + Target table + Target output columns + + + Sources + Referenced source tables + CTE and UNION resolved + + + Direct lineage + Per target output column + IDENTITY: source column + TRANSFORMATION: expression + AGGREGATION: aggregate + + + Indirect lineage + Dataset: JOIN, FILTER, + GROUP_BY, SORT + Output: WINDOW, CONDITIONAL + + + LineageInfo + Tables and columns + Expression dependencies + LineageContext + + + Plugin output + Stable dataset IDs + Column dependencies + Downstream schema + + + Query context + Query ID, SQL, user + Time and catalog details + + + + + + + + diff --git a/versioned_docs/version-4.x/data-governance/data-lineage.md b/versioned_docs/version-4.x/data-governance/data-lineage.md new file mode 100644 index 0000000000000..6b4bf02092ecc --- /dev/null +++ b/versioned_docs/version-4.x/data-governance/data-lineage.md @@ -0,0 +1,336 @@ +--- +{ + "title": "Data Lineage Technical Guide (4.x)", + "sidebar_label": "Data Lineage", + "language": "en", + "description": "Collect table-level and column-level lineage from supported Doris DML statements through an external LineagePlugin.", + "keywords": ["Apache Doris", "data lineage", "data governance", "LineagePlugin", "column-level lineage"] +} +--- + + + + +Data Lineage extracts table-level and column-level dependencies from supported Doris DML statements and sends them to external governance systems through a `LineagePlugin`. Doris is the lineage producer. It does not retain events, expose a lineage query API, or provide lineage visualization. + +> This capability is available in the 4.x release line starting with Apache Doris 4.0.6. + +## Capabilities and limitations + +The framework generates an event only after one of the following statements succeeds: + +| Supported statement | Event behavior | +| --- | --- | +| `INSERT INTO ... SELECT` | Generates one event after the Insert succeeds. | +| `INSERT OVERWRITE TABLE ... SELECT` | Generates one event after the Overwrite succeeds. | +| `CREATE TABLE AS SELECT` | Generates one event after the internal Insert succeeds. | + +The current implementation does not generate events for `SELECT`, `UPDATE`, `DELETE`, load jobs, `VALUES`-only inserts, or writes whose target is `__internal_schema`. Some `UPDATE` and `DELETE` execution paths internally reuse an Insert command, but the original command-type check excludes them from lineage event submission. + +:::caution Delivery guarantee + +Lineage delivery is asynchronous and best effort. A successful DML statement is not rolled back when event collection or plugin delivery fails. Events can be discarded when the queue is full, and Doris does not retry or persist them. + +::: + +## Technical design + +### Collection flow + +For a supported statement, Doris records the analyzed Nereids logical plan in an `afterAnalyze` planner hook. After the DML succeeds, Doris extracts lineage from that plan and submits the event to an FE-local queue. A single daemon worker evaluates every loaded plugin and dispatches the event to each plugin whose `eventFilter()` returns `true`. + +![Data lineage collection architecture: successful supported DML statements are analyzed by Nereids, extracted into LineageInfo, queued in FE, and delivered by a plugin to an external governance system.](/images/data-lineage/lineage-architecture.svg) + +Before extraction, Doris calls `eventFilter()` on the loaded plugins. If no plugin is willing to receive events, extraction is skipped. The worker evaluates `eventFilter()` again before dispatch. + +### Lineage content + +Each `LineageInfo` event contains table lineage, direct column lineage, two kinds of indirect lineage, and query context. The complete SQL example below establishes the source tables, target table, and result used by the following sections. + +#### Example SQL and result + +The current user must have privileges to create a database and tables, write data, and query data, and the FE must have loaded a lineage plugin. The example uses two source tables and creates a customer summary through a CTE, Join, filters, aggregation, a window function, a conditional expression, and sorting. + +```sql +CREATE DATABASE IF NOT EXISTS lineage_demo; +USE lineage_demo; + +CREATE TABLE lineage_orders ( + order_id BIGINT, + customer_id BIGINT, + region VARCHAR(16), + amount DECIMAL(18, 2), + status VARCHAR(16) +) +DUPLICATE KEY(order_id) +DISTRIBUTED BY HASH(order_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_customers ( + customer_id BIGINT, + customer_name VARCHAR(64), + customer_level VARCHAR(16) +) +DUPLICATE KEY(customer_id) +DISTRIBUTED BY HASH(customer_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +CREATE TABLE lineage_customer_summary ( + customer_id BIGINT, + region_label VARCHAR(16), + total_amount DECIMAL(18, 2), + customer_seq BIGINT, + region_group VARCHAR(16) +) +DUPLICATE KEY(customer_id) +DISTRIBUTED BY HASH(customer_id) BUCKETS 1 +PROPERTIES ("replication_num" = "1"); + +INSERT INTO lineage_customers VALUES + (1, 'Alice', 'VIP'), + (2, 'Bob', 'VIP'), + (3, 'Carol', 'REGULAR'); + +INSERT INTO lineage_orders VALUES + (101, 1, 'east', 100.00, 'PAID'), + (102, 1, 'east', 50.00, 'PAID'), + (103, 2, 'west', 80.00, 'PAID'), + (104, 2, 'west', 30.00, 'CANCELLED'), + (105, 3, 'north', 200.00, 'PAID'); + +INSERT INTO lineage_customer_summary +WITH customer_totals AS ( + SELECT + o.customer_id, + UPPER(o.region) AS region_label, + SUM(o.amount) AS total_amount + FROM lineage_orders o + JOIN lineage_customers c + ON o.customer_id = c.customer_id + WHERE o.status = 'PAID' + AND c.customer_level = 'VIP' + GROUP BY o.customer_id, UPPER(o.region) + HAVING SUM(o.amount) >= 50 +) +SELECT + customer_id, + region_label, + total_amount, + ROW_NUMBER() OVER (ORDER BY customer_id) AS customer_seq, + CASE + WHEN region_label = 'EAST' THEN 'CORE' + ELSE 'OTHER' + END AS region_group +FROM customer_totals +ORDER BY total_amount DESC; + +SELECT customer_id, region_label, total_amount, customer_seq, region_group +FROM lineage_customer_summary +ORDER BY customer_id; +``` + +The query returns: + +```text ++-------------+--------------+--------------+--------------+--------------+ +| customer_id | region_label | total_amount | customer_seq | region_group | ++-------------+--------------+--------------+--------------+--------------+ +| 1 | EAST | 150.00 | 1 | CORE | +| 2 | WEST | 80.00 | 2 | OTHER | ++-------------+--------------+--------------+--------------+--------------+ +``` + +The two `VALUES`-only inserts do not generate lineage events. The `INSERT INTO lineage_customer_summary ... SELECT` statement generates one `LineageInfo` event, whose contents are described below. + +#### Table lineage + +Table lineage records one target table and every source table scanned by the analyzed logical plan. The query that writes `lineage_customer_summary` scans both `lineage_orders` and `lineage_customers`, so both source tables have a table-level relationship with the target. A CTE is resolved to its underlying tables, and every `UNION` branch contributes to the source-table set. + +This example produces the following table lineage: + +| Source table | Target table | +| --- | --- | +| `lineage_orders` | `lineage_customer_summary` | +| `lineage_customers` | `lineage_customer_summary` | + +#### Direct column lineage + +Direct column lineage describes which source expression produces each target-column value. After resolving aliases and CTEs, Doris classifies expressions in this order: an expression containing an aggregate function is `AGGREGATION`; a plain source-column reference is `IDENTITY`; every other expression is `TRANSFORMATION`. + +| Type | Meaning | Example | +| --- | --- | --- | +| `IDENTITY` | The target value is taken directly from a source column without a function or operation. | `lineage_customer_summary.customer_id` comes directly from `lineage_orders.customer_id`. | +| `TRANSFORMATION` | A non-aggregate expression transforms the target value, such as arithmetic, a string function, a window function, or a conditional expression. | `region_label` comes from `UPPER(lineage_orders.region)`; `customer_seq` comes from `ROW_NUMBER()`. | +| `AGGREGATION` | The target value is produced by an expression that contains an aggregate function. It remains this type even when another expression wraps the aggregate. | `total_amount` comes from `SUM(lineage_orders.amount)`. | + +This example produces the following direct column lineage: + +| Target column | Source expression | Type | +| --- | --- | --- | +| `lineage_customer_summary.customer_id` | `lineage_orders.customer_id` | `IDENTITY` | +| `lineage_customer_summary.region_label` | `UPPER(lineage_orders.region)` | `TRANSFORMATION` | +| `lineage_customer_summary.total_amount` | `SUM(lineage_orders.amount)` | `AGGREGATION` | +| `lineage_customer_summary.customer_seq` | `ROW_NUMBER() OVER (ORDER BY lineage_orders.customer_id)` | `TRANSFORMATION` | +| `lineage_customer_summary.region_group` | `CASE WHEN UPPER(lineage_orders.region) = 'EAST' ... END` | `TRANSFORMATION` | + +#### Dataset indirect lineage + +Dataset indirect lineage records expressions that affect the complete result set without directly producing a target-column value. These dependencies are stored in an event-level map and can be applied to every target output column when needed. + +| Type | Meaning | Example | +| --- | --- | --- | +| `JOIN` | Join conditions determine which source rows can be combined. | `lineage_orders.customer_id = lineage_customers.customer_id`. | +| `FILTER` | `WHERE` or `HAVING` conditions determine which rows or groups enter the result. | `status = 'PAID'`, `customer_level = 'VIP'`, and `HAVING SUM(amount) >= 50`. | +| `GROUP_BY` | Grouping expressions determine the granularity of aggregate results. | Group by `customer_id` and `UPPER(region)`. | +| `SORT` | `ORDER BY` or TopN expressions determine result order. | Sort by `total_amount DESC`, which resolves to `SUM(amount)`. | + +This example produces the following dataset indirect lineage: + +| Type | Example source expression in the event | Effect | +| --- | --- | --- | +| `JOIN` | `lineage_orders.customer_id = lineage_customers.customer_id` | Determines how orders match customers. | +| `FILTER` | `status = 'PAID'`, `customer_level = 'VIP'`, and `SUM(amount) >= 50` | The `WHERE` and `HAVING` predicates jointly determine which rows and groups enter the result. | +| `GROUP_BY` | `lineage_orders.customer_id` and `UPPER(lineage_orders.region)` | Determine the aggregation granularity for each customer and region. | +| `SORT` | `total_amount DESC`, resolved to `SUM(lineage_orders.amount)` | Determines the order of the Insert query result. | + +#### Output indirect lineage + +Output indirect lineage is attached only to the affected target column and does not apply to the other output columns. + +| Type | Meaning | Example | +| --- | --- | --- | +| `WINDOW` | The `PARTITION BY` and `ORDER BY` inputs of a window function affect its output column. | `customer_seq` is produced by `ROW_NUMBER() OVER (ORDER BY customer_id)`, whose `WINDOW` dependency is `customer_id`. | +| `CONDITIONAL` | A `CASE` or `IF` condition, or the candidate expressions of `COALESCE`, affect its output column. | `region_group` is determined by `CASE WHEN region_label = 'EAST' ...`, whose `CONDITIONAL` dependency is that predicate. | + +The window or conditional expression itself still appears in direct column lineage. For example, the direct type of `customer_seq` is `TRANSFORMATION`, and it also has `WINDOW` indirect lineage. The direct type of `region_group` is also `TRANSFORMATION`, and it also has `CONDITIONAL` indirect lineage. + +This example produces the following output indirect lineage: + +| Target column | Type | Source expression in the event | Effect | +| --- | --- | --- | --- | +| `customer_seq` | `WINDOW` | `lineage_orders.customer_id` | Serves as the window ordering input to `ROW_NUMBER()` and affects only `customer_seq`. | +| `region_group` | `CONDITIONAL` | `UPPER(lineage_orders.region) = 'EAST'` | Serves as the `CASE WHEN` condition and affects only `region_group`. | + +The other three target columns have no output indirect lineage. Dataset-level `JOIN`, `FILTER`, `GROUP_BY`, and `SORT` dependencies still affect the complete result set. + +#### Query context + +Query context identifies who produced the lineage event, in which session, and through which DML statement. + +| Field | Example or meaning | +| --- | --- | +| Source Command | `InsertIntoTableCommand`, indicating that the event comes from an Insert. | +| Query ID and original SQL | The Query ID and the complete `INSERT INTO lineage_customer_summary ...` text. | +| User and client IP | For example, `lineage_user` and `192.0.2.10`. | +| Session database and catalog | `lineage_demo` and `internal` in this example. They describe session context and do not necessarily identify the target-table location. | +| Execution state | A successful DML statement normally records `OK`. | +| Timestamp and duration | The event creation time and the milliseconds from query start to event creation. The value is `-1` when unavailable. | +| External catalog properties | Non-sensitive properties from external catalogs referenced by the query. Passwords, keys, and hidden properties are removed. The map is empty when only the `internal` catalog is used. | + +For this example, the original SQL is the complex Insert above, the Source Command is `InsertIntoTableCommand`, the session database is `lineage_demo`, the session catalog is `internal`, and the successful state is `OK`. The Query ID, user, client IP, timestamp, and duration use values from the actual execution context. Because the example uses only the Internal Catalog, the external catalog properties map is empty. + +Confirm in the plugin's downstream system that the event was received. During initial deployment, also check `fe.log` for `Loaded lineage plugin`, plugin-specific errors, and queue-full warnings. + +![LineageInfo event model: target and source tables feed direct and indirect lineage, which are combined with query context before a plugin converts the event for a downstream system.](/images/data-lineage/lineage-event-model.svg) + +The extractor resolves CTE producer expressions and expands `UNION` branches before the event is delivered. A downstream plugin must convert Doris Java objects such as `TableIf`, `SlotReference`, and `Expression` to stable identifiers or its own event schema. + +### Event processing behavior + +`lineage_event_queue_size` sets the maximum number of lineage events waiting in each FE-local queue. It counts events, not bytes. When the queue is full, the new event is discarded and the DML continues normally. The worker invokes plugins serially, so a slow plugin delays later events and can cause drops under sustained load. Exceptions thrown by a plugin are logged and do not stop the worker or affect the DML. + +Plugin discovery and initialization occur only when FE starts. Dynamic reload and unload are not supported. + +## Configure and use + +### Prerequisites + +1. Prepare an external lineage plugin. Doris does not ship a built-in sink. For the SPI contract, JAR packaging, and a complete minimal plugin, see [Data Lineage Plugin Development](/community/developer-guide/data-lineage-plugin-development). +2. Copy the plugin JAR and any required third-party dependency JARs to every FE. +3. If the plugin sends events to an external governance service, make sure that service is reachable from every FE that can execute the DML. + +### Deploy the plugin + +`FE_HOME` denotes a single FE installation directory that contains `bin/`, `conf/`, and `lib/`. Use the following layout on every FE. The loader scans only direct child directories under `lineage/`, plus JARs in each plugin directory and its `lib/` directory. + +```text +$FE_HOME/plugins/ +└── lineage/ + └── example-lineage/ + ├── example-lineage.jar + └── lib/ + └── downstream-client.jar +``` + +Configure `fe.conf` on every FE. Replace `example-lineage` with the value returned by the plugin factory's `name()` method: + +```text +plugin_dir = /opt/apache-doris/fe/plugins +activate_lineage_plugin = example-lineage +lineage_event_queue_size = 50000 +``` + +| Configuration | Type and default | Required | Description | +| --- | --- | --- | --- | +| `plugin_dir` | String; `$FE_HOME/plugins` | No | Plugin root directory. The lineage loader scans its direct children under `lineage/`. Omit this setting when using the default directory. | +| `activate_lineage_plugin` | String array; empty | Explicit configuration recommended | Comma-separated Factory names to instantiate. An empty value disables name filtering and instantiates all discovered factories. | +| `lineage_event_queue_size` | Positive integer; `50000` | No | Maximum pending lineage events on the current FE. New events are discarded when the queue is full. | + +All three settings are configured in `$FE_HOME/conf/fe.conf` on each FE and apply only to that FE process. They are read only at FE startup, so restart the corresponding FE after changing them. + +#### Activate plugins + +Configure `activate_lineage_plugin` in `$FE_HOME/conf/fe.conf` on every FE. It controls which discovered factories are instantiated. It does not discover JARs or control queue capacity. Each name is case-sensitive and must exactly match `LineagePluginFactory.name()`. The `LineagePlugin.name()` implementation should return the same name. The plugin directory name is not used for matching. Separate multiple plugin names with commas. The FE configuration parser trims the spaces around each comma-separated item, but name matching remains case-sensitive: + +```text +activate_lineage_plugin = example-lineage,governance-lineage +``` + +At startup, FE discovers built-in and external factories, then creates instances only for the configured names. Each FE reads its own configuration, so every FE that can execute DML must use the same plugin set. Restart FE after a change. This setting cannot be changed dynamically through `ADMIN SET FRONTEND CONFIG`. + +:::caution Empty value behavior + +In the current implementation, an empty `activate_lineage_plugin` value does not disable lineage plugins. It skips name filtering and instantiates every discovered factory. Explicitly list the plugins to enable, and do not use an empty value as a disable switch. After a plugin is loaded, `eventFilter()` still decides whether it receives an event before extraction and before dispatch. + +::: + +#### Configure the event queue + +Configure `lineage_event_queue_size` in `$FE_HOME/conf/fe.conf` on every FE. It must be a positive integer. Each FE that can execute DML has an independent queue and must be configured separately. + +This is an FE startup setting and cannot be changed dynamically through `ADMIN SET FRONTEND CONFIG`. Restart the corresponding FE after changing it. Increasing the value increases pending-event capacity and FE memory usage. It does not add consumer threads or increase plugin processing throughput. When events are missing, first search `fe.log` for `the lineage event queue is full` and reduce latency in the plugin's `exec()` method. Then size the queue according to peak backlog and available FE memory. + +Restart every FE after changing other plugin settings or replacing plugin JARs. Plugin configuration and directories are read only at FE startup. + +## Operate and troubleshoot + +### No event after a write + +Common causes include an unsupported statement, a failed DML statement, no plugin loaded on the current FE, or `eventFilter()` returning `false`. + +Check in this order: + +1. Confirm that the statement is a successful `INSERT INTO ... SELECT`, `INSERT OVERWRITE TABLE ... SELECT`, or `CREATE TABLE AS SELECT`. A `VALUES`-only Insert does not generate an event. +2. Confirm that the FE that executed the DML has configured and loaded the plugin. Search `fe.log` for `Loaded lineage plugin`. +3. Confirm that the plugin's `eventFilter()` returns `true` in both the query thread and the worker thread. +4. Check plugin logs and the downstream service to confirm that processing did not fail outside FE. + +### Plugin not loaded + +Confirm that the plugin directory contains a JAR and that the JAR contains `META-INF/services/org.apache.doris.nereids.lineage.LineagePluginFactory`. Each plugin directory can expose only one Factory. Factory names must be globally unique and must match `activate_lineage_plugin`. Restart the corresponding FE after correcting the directory, JAR, or configuration. + +### Missing events under high load + +Search `fe.log` for `the lineage event queue is full` and plugin exceptions. If the queue is full, first reduce synchronous blocking in `exec()`, and make downstream delivery idempotent with bounded retries. Then evaluate whether to increase `lineage_event_queue_size`. Before increasing it, account for FE memory because event sizes vary by SQL statement. + +Also confirm that every FE that can execute DML has the same plugin and configuration. Each FE uses an independent local queue; plugin and queue state are not synchronized across FEs. + +### Plugin update did not take effect + +Plugin directories are scanned only when FE starts. Dynamic reload and unload are not supported. Restart every relevant FE after replacing plugin JARs, dependencies, or configuration, and check `Loaded lineage plugin` again. + +Use the Query ID and target table to construct downstream event identifiers. This allows the governance system to deduplicate events even when the plugin implements retries outside the framework. + +## Further reading + +- [Data Lineage Plugin Development](/community/developer-guide/data-lineage-plugin-development) diff --git a/versioned_sidebars/version-4.x-sidebars.json b/versioned_sidebars/version-4.x-sidebars.json index 1e31cb0f28e01..72acf29c800ad 100644 --- a/versioned_sidebars/version-4.x-sidebars.json +++ b/versioned_sidebars/version-4.x-sidebars.json @@ -883,6 +883,13 @@ ] } ] + }, + { + "type": "category", + "label": "Data Governance", + "items": [ + "data-governance/data-lineage" + ] } ] },