Spark 3.5: retain tracked-column stats for Column-Value Lineage (CVL) by att10 · Pull Request #278 · linkedin/iceberg

att10 · 2026-09-17T23:49:08Z

What

Enables per-file min/max bound retention on the Spark 3.5 / Iceberg 1.5 batch read path so downstream Column-Value Lineage (CVL) can compute value bounds for the tracked column(s). Previously the batch scan dropped column stats, so the CVL adapter (li-openhouse) had no bounds to read; it cannot re-add stats after planning, so the retention must be enabled here in the Iceberg Spark integration.

How

SparkSQLProperties — new session flag spark.lineage.columnValues.enabled (default on).
SparkReadConf — columnValueLineageEnabled() accessor.
SparkScanBuilder:
- Table-property constants lineage.columnValues.enabled (per-table kill switch, default on) and lineage.columnValues.columns (default datepartition).
- lineageStatsColumns(...) helper: returns empty when disabled by the session flag or the per-table kill switch; otherwise resolves the configured columns honoring the scan's caseSensitive flag and returns their canonical schema names.
- buildBatchScan(...) calls scan.includeColumnStats(statsColumns) (requested-columns overload → bounded overhead) inside a fail-safe block, so any error skips retention without failing the query.

Default behavior tracks only datepartition; overridable per table.

Scope

Batch query scan only. Incremental / changelog / copy-on-write / aggregate-pushdown paths and write-path CVL are unchanged. No changes to li-openhouse or the engine fork (separate repos).

Testing

New TestColumnValueLineageStats (7 cases): default retention, non-tracked columns dropped, session flag off, per-table kill switch, columns override, missing-column ignored, and case-insensitive resolution. All pass on the iceberg-spark-3.5_2.12 module.

…#1) * Spark 3.5: retain tracked-column stats for Column-Value Lineage Enable per-file min/max bound retention on the batch read path so downstream Column-Value Lineage (CVL) can compute value bounds for the tracked column(s). Gated by the session flag spark.lineage.columnValues.enabled and the per-table lineage.columnValues.enabled kill switch; tracked columns come from lineage.columnValues.columns (default datepartition) and use the requested-columns includeColumnStats overload to bound overhead. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Resolve tracked lineage columns using the scan's case sensitivity Honor the scan's caseSensitive flag when resolving lineage.columnValues columns and pass the schema's canonical field name to includeColumnStats so case-insensitive scans match differently-cased column names. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Address review: fail-safe stats retention and hourly test partitions Wrap Column-Value Lineage column-stats retention in a fail-safe block so any error skips retention without failing the query, and switch test datepartition values to the hourly '2024-01-01-00' format. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

yangyangv2 · 2026-09-18T01:56:57Z

+    try {
+      List<String> statsColumns = lineageStatsColumns(expectedSchema);
+      if (!statsColumns.isEmpty()) {
+        scan = scan.includeColumnStats(statsColumns);


the collected statsColumns are case sensitive (from the config), would the is the scanner case sensitive or insensitive?

e.g. config = Datepartition, column = datepartition, would they match?

They should match when caseSensitive=false which I believe is the spark default. By the time we call includeColumnStats(statsColumns), the names are already canonicalized. In lineageStatsColumns we resolve each configured name honoring the scan's caseSensitive flag:

Types.NestedField field = caseSensitive ? expectedSchema.findField(name) : expectedSchema.caseInsensitiveFindField(name); if (field != null) { statsColumns.add(field.name()); // canonical, exact-case schema name }

caseSensitive=false (Spark default): config Datepartition → caseInsensitiveFindField → resolves to datepartition, and we pass the canonical field.name() downstream, so the later case-sensitive findField in includeColumnStats matches.

caseSensitive=true: matching is exact, so Datepartition ≠ datepartition → findField returns null → the null guard skips it (no NPE, no bounds).

yangyangv2 · 2026-09-18T02:00:38Z

    scan = configureSplitPlanning(scan);

+    try {
+      List<String> statsColumns = lineageStatsColumns(expectedSchema);


Would it be worth of moving this configuration out, e.g. as a static across the application lifecycle?

lineage.columnValues.enabled and lineage.columnValues.columns are per-table and read from table.properties(), so two tables in the same app can have different settings.

spark.lineage.columnValues.enabled is per-session (Spark conf), so it can change between sessions/queries.

So I think they have to be resolved per scan? Let me know if I'm misunderstanding your ask though

you're right about the table properties. lineage.columnValues.columns, every scan needs to resolve it.

The session properties spark.lineage.columnValues.enabled can be at the session level instead of scan-level.

yangyangv2 · 2026-09-18T02:03:20Z

+   * unaffected.
+   */
+  private List<String> lineageStatsColumns(Schema expectedSchema) {
+    if (!readConf.columnValueLineageEnabled()) {


spark can also pass session config to disable the column value lineage connection, can you check if it can be also checked here?

e.g. spark.lineage.columnValues.enabled=false

https://github.com/linkedin-multiproduct/spark-grid-template/pull/70/changes#diff-c42ae28bed8f629bb69576fb75ee2b9dfa378ff6055febffcc79d66bf04a536aL609

Yup columnValueLineageEnabled() checks the value of spark.lineage.columnValues.enabled

jiang95-dev

Lineage is a concept in the catalog not table format. Including lineage implementation at iceberg level is not right.

Let's redesign like this: In iceberg repo we just accept columns as input and build the Scan using the columns. Then in li-openhouse repo we build the lineage adaptor, and pass these lineage specific columns to the builder. We should also be careful to call iceberg-1.2 and iceberg-1.5 in a compatible way.

…e-specific Remove lineage-specific config (spark.lineage.columnValues.enabled and the lineage.columnValues.* table properties) from the iceberg Spark integration so the template repo stays lineage-agnostic. Replace it with a generic report-column-stats read option / session config listing the columns whose per-file stats (min/max bounds) should be retained on planned batch tasks. Callers such as li-openhouse own all lineage semantics and pass the resolved columns in via this generic knob. Case-insensitive resolution, canonical-name handling, and fail-safe behavior are preserved. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Replace the report-column-stats read option / session config with a public version-agnostic control method SparkScanBuilder.includeColumnStats(Collection), mirroring the 1.2 API in linkedin#279. The consumer (e.g. li-openhouse) passes the columns directly; on 1.5 they map to the per-column Scan.includeColumnStats(Collection). Names are resolved against the scan schema honoring case sensitivity (canonical names, unknown columns ignored) and applied inside the existing fail-safe block. No config or lineage naming remains in the iceberg repo. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

github-actions Bot added the SPARK label Sep 17, 2026

Apply spotless formatting to SparkScanBuilder javadoc

8edb1fb

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

yangyangv2 reviewed Sep 18, 2026

View reviewed changes

jiang95-dev requested changes Sep 18, 2026

View reviewed changes

yangyangv2 mentioned this pull request Sep 21, 2026

Spark 3.1: add includeColumnStats(Collection<String>) control API on SparkScanBuilder #279

Open

yangyangv2 approved these changes Sep 21, 2026

View reviewed changes

Navigation Menu

Spark 3.5: retain tracked-column stats for Column-Value Lineage (CVL) - #278

Spark 3.5: retain tracked-column stats for Column-Value Lineage (CVL)#278
att10 wants to merge 4 commits into
linkedin:openhouse-1.5.2from
att10:cvl-iceberg-1.5-upstream

att10 commented Sep 17, 2026

Uh oh!

yangyangv2 Sep 18, 2026

Uh oh!

att10 Sep 18, 2026 •

edited

Loading

Uh oh!

yangyangv2 Sep 18, 2026 •

edited

Loading

Uh oh!

att10 Sep 18, 2026

Uh oh!

yangyangv2 Sep 18, 2026

Uh oh!

yangyangv2 Sep 18, 2026

Uh oh!

att10 Sep 18, 2026

Uh oh!

jiang95-dev left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

att10 commented Sep 17, 2026

What

How

Scope

Testing

Uh oh!

yangyangv2 Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

att10 Sep 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yangyangv2 Sep 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

att10 Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

yangyangv2 Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

yangyangv2 Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

att10 Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

jiang95-dev left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

att10 Sep 18, 2026 •

edited

Loading

yangyangv2 Sep 18, 2026 •

edited

Loading