Skip to content

[flink] Add data evolution delete action#8716

Open
wwj6591812 wants to merge 1 commit into
apache:masterfrom
wwj6591812:agent/data-evolution-delete-action
Open

[flink] Add data evolution delete action#8716
wwj6591812 wants to merge 1 commit into
apache:masterfrom
wwj6591812:agent/data-evolution-delete-action

Conversation

@wwj6591812

@wwj6591812 wwj6591812 commented Jul 18, 2026

Copy link
Copy Markdown
Contributor

Purpose

Related to #8322.

This PR adds a bounded Flink Action named data_evolution_delete for logically deleting rows from non-primary-key Data Evolution append tables by writing deletion vectors.

The existing Flink delete Action emits DELETE changelog records and does not provide a deletion-vector path for non-primary-key Data Evolution append tables. Rewriting data files for random deletion is particularly expensive for Data Evolution tables with dedicated BLOB files. This Action provides a bounded cleanup path for retention policies, invalid data cleanup, and cache eviction.

What the Action does

  • validates that the target is a non-primary-key, bucket-unaware table with row tracking, Data Evolution, and deletion vectors enabled;
  • evaluates a Flink SQL predicate against a fixed snapshot of the target row-tracking system table;
  • supports repeatable --source_sql statements, executed in order, to register bounded external sources and views (for example JDBC/Hologres candidates) that are referenced by a subquery in --where;
  • maps matched global row IDs to Data Evolution anchor files and local row positions;
  • aggregates row positions in parallel per anchor file, then shuffles the compressed deletion vectors by a rewrite-group ownership key;
  • assigns all anchors backed by one existing DV index file to exactly one writer, while anchors without an existing DV are distributed over stable writer shards;
  • scans only old DV index files owned by the current writer, and skips old-index scanning completely for new-only writers;
  • merges new deletion vectors with existing deletion vectors and commits one index-only DELETE snapshot without rewriting data or dedicated BLOB files;
  • uses strict fixed-snapshot conflict checks so concurrent APPEND, COMPACT, OVERWRITE, or delete commits fail instead of silently overwriting DV state;
  • does not echo external source SQL or its original exception text on source-registration failure, because connector DDL may contain credentials.

Using an external subquery avoids copying a large Hologres/JDBC candidate set into an intermediate Paimon table.

Example

<FLINK_HOME>/bin/flink run   /path/to/paimon-flink-action-<version>.jar   data_evolution_delete   --warehouse <warehouse-path>   --database <database-name>   --table <table-name>   --source_sql "CREATE TEMPORARY TABLE deletion_candidates (
      url STRING
  ) WITH (
      'connector' = 'jdbc',
      'url' = '<jdbc-url>',
      'table-name' = '<candidate-table>'
  )"   --where "url IN (SELECT url FROM deletion_candidates)"   --sink_parallelism 8

The external source must be bounded.

Ownership and operational constraints

An existing physical DV index file is the atomic rewrite group because it may contain deletion vectors for several anchor files. Splitting one old index file across writers could produce competing replacements, so it always has a single owner. Independent old index files and new-anchor shards can be written in parallel.

This Action performs a logical delete. Physical data and BLOB reclamation still requires Data Evolution compaction with data-evolution.compaction.rewrite-row-ids=true, followed by snapshot expiration.

Run at most one mutation job against the same target table at a time. The fixed base snapshot must remain retained until the Action finishes. Split very large deletes into bounded batches to limit row-tracking planning, deletion-vector memory, and external-source scan size.

This PR intentionally adds the Action only; a Flink Procedure can be added separately.

Tests

mvn -pl paimon-core,paimon-flink/paimon-flink-common   -Pflink1 -DskipTests spotless:check

mvn -pl paimon-flink/paimon-flink-common -am -Pflink1   -Dtest=DataEvolutionDeleteActionITCase,ProcedureTest   -Dsurefire.failIfNoSpecifiedTests=false   -Dspotless.check.skip=true test

The targeted run passes 12 tests with 0 failures, errors, or skips:

  • 10 DataEvolutionDeleteAction integration tests covering cross-partition deletion, merging existing DVs, BLOB rows without data/BLOB rewrite, no-match behavior, bounded external candidates, ordered source SQL, source-SQL failure redaction and no-snapshot behavior, parallel rewrite groups on an unpartitioned table, rewrite-group ownership, fixed-snapshot conflicts, required table options, and required where filters;
  • 2 Procedure tests validating Action/Procedure coverage and named arguments.

Spotless, Checkstyle, and git diff --check also pass. The full suite will be exercised by CI.

@wwj6591812
wwj6591812 marked this pull request as ready for review July 18, 2026 13:25
@wwj6591812
wwj6591812 force-pushed the agent/data-evolution-delete-action branch from dfd5371 to 3188506 Compare July 19, 2026 08:20
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant