[flink] Add data evolution delete action

[wwj6591812]

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.

[steFaiz]

Thanks for this PR! I'm thinking that do we need to implement independent actions for both PK, Append and DataEvolution tables? If so, we may need to trigger different procedures according to the table type.

Can we just implement one delete action but solve different table type independently just like spark?
org.apache.paimon.spark.commands.DeleteFromPaimonTableCommand

Now we could just mark PK and append-only tables as unsupported.

[JingsongLi]

+1 to @steFaiz

[wwj6591812]

Thanks for this PR! I'm thinking that do we need to implement independent actions for both PK, Append and DataEvolution tables? If so, we may need to trigger different procedures according to the table type.

Can we just implement one delete action but solve different table type independently just like spark? org.apache.paimon.spark.commands.DeleteFromPaimonTableCommand

Now we could just mark PK and append-only tables as unsupported.

Thanks for the suggestion. I have consolidated the implementation into the existing delete action, so users now have a single entry point and the action dispatches internally according to the target table type, similar to Spark's DeleteFromPaimonTableCommand. The existing primary-key delete path is retained for backward compatibility, Data Evolution append tables use the deletion-vector implementation introduced by this PR, and regular append-only tables remain explicitly unsupported. I also removed the separate data_evolution_delete factory and identifier, moved its options into DeleteActionFactory, and updated the documentation and tests, including the bounded external source filtering case. All 15 targeted PK, Data Evolution, and procedure tests pass.

@steFaiz @JingsongLi
Please CC, Thx.