yidawang-shopify opened a new pull request, #17668:
URL: https://github.com/apache/iceberg/pull/17668
## What
Makes the Flink connector report **source → sink table lineage** through
Flink's FLIP-314
`JobStatusChangedListener`, so a listener can resolve which Iceberg tables a
job reads and writes
without anyone hand-declaring the edge.
`IcebergSource` and `IcebergSink` implement `LineageVertexProvider`. A new
`IcebergLineageUtil`
builds the `LineageDataset` those vertices publish.
## Why the diff touches the table source and sink
Flink only finds a lineage vertex if it can reach the `Source`/`Sink`
**object**, and that depends
on which `RuntimeProvider` the connector returns:
| Provider | What the connector hands over | Planner can see the object? |
|---|---|---|
| `DataStreamScanProvider` / `DataStreamSinkProvider` | a callback that
builds a topology | No — the object is a local inside the lambda |
| `SourceProvider` / `SinkV2Provider` | the object itself | Yes |
`LineageGraphUtils#processSource` reads the vertex off
`SourceTransformation` and `processSink` off
`SinkTransformation`, and those transformations only carry a vertex when the
planner constructed
them itself — i.e. from a declarative provider.
So `IcebergTableSource` and `IcebergTableSink` now hand over
`SourceProvider.of(source)` /
`SinkV2Provider.of(sink)` instead of a callback. Everything else follows
from that: the source
infers its own parallelism now that it no longer goes through `buildStream`,
and the sink gains a
`build()` that does not attach a topology.
## What the dataset carries
Iceberg's own vocabulary only — catalog, namespace, table, and the catalog's
`uri`/`warehouse` —
plus the `prefix` a REST catalog's server assigned to this client. Composing
a vendor-specific
fully-qualified name is left to the listener, which is the only party with
the deployment context.
Namespaces are reported verbatim for the same reason: Iceberg permits any
depth, and flattening one
to fit some vendor's scheme isn't the connector's decision.
The prefix is read from the catalog rather than derived from `warehouse`,
which is wrong for any
catalog not backed by exactly one bucket. It is the handle used for every
`/v1/<prefix>/...` call,
arrives in `GET /v1/config`, and is merged into the catalog's properties at
initialization.
Coordinates live in a `DatasetConfigFacet` rather than in
`LineageDataset#name()` because on the SQL
path the planner wraps the dataset in `TableLineageDatasetImpl`, which
overwrites `name()` with the
Flink object identifier; `namespace()` and `facets()` survive.
## Cost on the submission path
Because only a live catalog knows the prefix, resolving it inside
`IcebergLineageUtil` would cost a
catalog initialization per table at job submission — and more than once per
table, since Flink asks
a connector for its lineage vertex both when it extracts the dataset and
when it constructs the
transformation. So the source and sink builders read the prefix from the
catalog they already opened
to load the table and pass it in, and each vertex resolves its datasets
once. An empty prefix records
that a live catalog answered "none", which keeps that answer from being
re-derived on every
submission.
Lineage is best-effort observability: every path yields no dataset rather
than throwing, so a table
whose coordinates can't be resolved never fails the job.
## Configuration
| | Flag | Default | Lineage |
|---|---|---|---|
| Source | `table.exec.iceberg.use-flip27-source` | `true` | works as-is |
| Sink | `table.exec.iceberg.use-v2-sink` | `false` | must opt in |
| Sink | `table.exec.uid.generation` | `PLAN_ONLY` | must be `ALWAYS` |
Source lineage needs no configuration change. Sink lineage needs both sink
settings; with either
missing the sink keeps the existing callback path and logs at INFO why.
DataStream API users need no
flags on either end, since they construct `IcebergSource`/`IcebergSink`
directly.
## Scope
`flink/v2.1` only, matching how recent Flink changes have landed. Happy to
follow up with backports
to `v2.0`/`v1.20` if reviewers would prefer them.
## Testing
- `TestIcebergLineageUtil` — facet contents across REST/Hadoop/non-catalog
loaders, and that a
captured prefix is reported without opening a catalog.
- `TestIcebergSqlLineage` — end-to-end `LineageGraph` on the SQL path via a
registered
`JobStatusChangedListener`.
- `TestIcebergSourceInferParallelism` — the source's own parallelism
inference.
## AI disclosure
This change was developed with AI assistance (Claude), reflected in the
commit's `Co-Authored-By`
trailer. I've reviewed the implementation end-to-end and verified the REST
`prefix` behaviour
against a live REST catalog rather than only a fixture. Open question I'd
value a reviewer's opinion
on: whether `IcebergLineageUtil`'s prefix-capture entry points are the right
public surface, or
whether they should be package-private with the capture folded into the
builders.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
