This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 9a629fd8293e feat(agent): add first mutable Flink SQL COW sink path to
Hudi Architect (#20281)
9a629fd8293e is described below
commit 9a629fd8293ef62f5f036038c58d2eb5e8d1cb04
Author: fhan <[email protected]>
AuthorDate: Tue Oct 13 10:44:50 2026 +0800
feat(agent): add first mutable Flink SQL COW sink path to Hudi Architect
(#20281)
* feat(agent): add first mutable Flink SQL COW sink path to Hudi Architect
* fix(agent): align mutable Flink replay and ordering validation
---------
Co-authored-by: fhan <[email protected]>
---
.github/workflows/agent_gateway_ci.yml | 3 +
.../fixtures/hudi-architect-flink-1.20/pom.xml | 8 +-
.../architect/TestFlinkArchitectSqlFixtures.java | 118 ++++++-
hudi-agent-gateway/skills/hudi-architect/README.md | 17 +-
.../skills/hudi-architect/RUNBOOK.md | 17 +-
hudi-agent-gateway/skills/hudi-architect/SKILL.md | 19 +-
.../flink-1.20-hudi-1.2.0-capabilities.md | 32 +-
.../flink-1.20-hudi-1.2.0-capabilities.toml | 171 +++++++++-
.../references/flink-config-templates.md | 57 +++-
.../references/flink-decision-overrides.md | 53 +++-
.../references/flink-question-flow.md | 50 ++-
.../hudi-architect/references/flink-warnings.md | 99 +++++-
.../hudi-architect/validate_flink_capabilities.py | 115 ++++++-
.../skills/hudi-architect/validate_flink_design.py | 344 ++++++++++++++++++---
.../hudi_architect/flink_pr1_scenarios.toml | 9 +-
.../hudi_architect/flink_pr3/mutable_cow.json | 66 ++++
.../hudi_architect/flink_pr3/mutable_cow.sql | 55 ++++
.../tests/test_hudi_architect_skill.py | 288 ++++++++++++++++-
18 files changed, 1363 insertions(+), 158 deletions(-)
diff --git a/.github/workflows/agent_gateway_ci.yml
b/.github/workflows/agent_gateway_ci.yml
index 6d6e5f35e613..aafccf8b6936 100644
--- a/.github/workflows/agent_gateway_ci.yml
+++ b/.github/workflows/agent_gateway_ci.yml
@@ -71,6 +71,9 @@ jobs:
python skills/hudi-architect/validate_flink_design.py \
--input tests/fixtures/hudi_architect/flink_pr2/stable_key.json \
--emit-sql > /dev/null
+ python skills/hudi-architect/validate_flink_design.py \
+ --input tests/fixtures/hudi_architect/flink_pr3/mutable_cow.json \
+ --emit-sql > /dev/null
- name: Test
run: python -m pytest tests -q
- name: Lint
diff --git a/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/pom.xml
b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/pom.xml
index c3226c49fb1f..0714ab8d6fbf 100644
--- a/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/pom.xml
+++ b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/pom.xml
@@ -26,7 +26,7 @@
<packaging>jar</packaging>
<name>Hudi Architect Flink 1.20 fixtures</name>
- <description>Pinned factory and planner fixtures for the first executable
Architect path.</description>
+ <description>Pinned factory and planner fixtures for bounded executable
Architect paths.</description>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@@ -82,6 +82,12 @@
<include>*.sql</include>
</includes>
</testResource>
+ <testResource>
+
<directory>${project.basedir}/../../tests/fixtures/hudi_architect/flink_pr3</directory>
+ <includes>
+ <include>*.sql</include>
+ </includes>
+ </testResource>
</testResources>
<plugins>
<plugin>
diff --git
a/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectSqlFixtures.java
b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectSqlFixtures.java
index a6c008620b5b..89ef68792fb7 100644
---
a/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectSqlFixtures.java
+++
b/hudi-agent-gateway/fixtures/hudi-architect-flink-1.20/src/test/java/org/apache/hudi/agent/architect/TestFlinkArchitectSqlFixtures.java
@@ -23,6 +23,7 @@ import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.sink.bulk.RowDataKeyGen;
import org.apache.hudi.table.HoodieTableFactory;
+import org.apache.hudi.util.ChangelogModes;
import org.apache.hudi.util.DataTypeUtils;
import org.apache.hudi.util.HoodieSchemaConverter;
@@ -46,6 +47,7 @@ import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.TimeType;
import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
import org.junit.jupiter.api.Test;
import java.io.IOException;
@@ -56,6 +58,7 @@ import java.sql.Date;
import java.sql.Timestamp;
import java.time.Duration;
import java.util.Collections;
+import java.util.List;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -64,11 +67,12 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
- * Factory and planner fixtures for the bounded Hudi Architect PR2 path.
+ * Factory and planner fixtures for the bounded Hudi Architect Flink SQL paths.
*
* <p>The fixture depends on the released Hudi 1.2.0 Flink 1.20 bundle and
Flink 1.20.1 rather
* than the enclosing checkout. It plans the exact SQL golden files checked by
the Python
@@ -120,6 +124,61 @@ class TestFlinkArchitectSqlFixtures {
assertTrue(plan.contains("changelogMode=[I]"), plan);
}
+ @Test
+ void testMutableCowSinkAndUpsertPlan() throws Exception {
+ TestContext context = newTestContext();
+ StreamTableEnvironment tableEnv = context.tableEnv;
+ registerMutableSource(context);
+
+ SqlFixture fixture = SqlFixture.load("mutable_cow.sql");
+ applyRuntimeStatements(tableEnv, fixture.runtimeSql);
+ tableEnv.executeSql(fixture.tableDdl);
+ String plan = tableEnv.explainSql(
+ fixture.insertSql,
+ ExplainDetail.CHANGELOG_MODE,
+ ExplainDetail.JSON_EXECUTION_PLAN);
+
+ assertPinnedArtifacts();
+ assertTrue(plan.contains("orders_mutable_hudi"), plan);
+ assertTrue(plan.contains("stream_write:
default_database.orders_mutable_hudi"), plan);
+ assertTrue(plan.contains("changelogMode=[I,UA,D]"), plan);
+ assertTrue(plan.contains("index_bootstrap"), plan);
+ assertEquals(
+ Set.of(RowKind.INSERT, RowKind.UPDATE_AFTER, RowKind.DELETE),
+ ChangelogModes.UPSERT.getContainedKinds());
+ assertFalse(ChangelogModes.UPSERT.contains(RowKind.UPDATE_BEFORE));
+ }
+
+ @Test
+ void testPinnedMutableOrderingConfiguration() {
+ Configuration conf = mutableConfiguration();
+
+ assertEquals("hoodie.write.record.merge.mode",
FlinkOptions.RECORD_MERGE_MODE.key());
+ assertEquals("event_version", conf.get(FlinkOptions.ORDERING_FIELDS));
+ assertEquals("EVENT_TIME_ORDERING",
conf.get(FlinkOptions.RECORD_MERGE_MODE));
+ assertFalse(conf.get(FlinkOptions.CHANGELOG_ENABLED));
+
+ conf.set(FlinkOptions.ORDERING_FIELDS, FlinkOptions.NO_PRE_COMBINE);
+ assertEquals("no_precombine", FlinkOptions.NO_PRE_COMBINE);
+ assertNull(OptionsResolver.getOrderingFieldsStr(conf));
+ }
+
+ @Test
+ void testPinnedMutableIndexConfiguration() {
+ Configuration conf = mutableConfiguration();
+
+ assertEquals("FLINK_STATE", conf.get(FlinkOptions.INDEX_TYPE));
+ assertTrue(conf.get(FlinkOptions.INDEX_GLOBAL_ENABLED));
+ assertEquals(0.0, conf.get(FlinkOptions.INDEX_STATE_TTL));
+ }
+
+ @Test
+ void testPinnedMutableIndexBootstrapConfiguration() {
+ Configuration conf = mutableConfiguration();
+
+ assertTrue(conf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED));
+ }
+
@Test
void testPinnedFactorySkipsMissingRecordKeyCheckInAppendMode() {
TestContext context = newTestContext();
@@ -359,6 +418,20 @@ class TestFlinkArchitectSqlFixtures {
return (RowDataKeyGen) factory.invoke(null, conf, rowType);
}
+ private static Configuration mutableConfiguration() {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_COPY_ON_WRITE);
+ conf.set(FlinkOptions.OPERATION, "upsert");
+ conf.set(FlinkOptions.ORDERING_FIELDS, "event_version");
+ conf.set(FlinkOptions.RECORD_MERGE_MODE, "EVENT_TIME_ORDERING");
+ conf.set(FlinkOptions.INDEX_TYPE, "FLINK_STATE");
+ conf.set(FlinkOptions.INDEX_GLOBAL_ENABLED, true);
+ conf.set(FlinkOptions.INDEX_STATE_TTL, 0.0);
+ conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true);
+ conf.set(FlinkOptions.CHANGELOG_ENABLED, false);
+ return conf;
+ }
+
private static TestContext newTestContext() {
StreamExecutionEnvironment environment =
StreamExecutionEnvironment.getExecutionEnvironment();
environment.enableCheckpointing(60_000L);
@@ -404,6 +477,49 @@ class TestFlinkArchitectSqlFixtures {
registerInsertOnlyView(context.tableEnv, "events_source", stream, schema);
}
+ private static void registerMutableSource(TestContext context) {
+ Row insert = Row.ofKind(
+ RowKind.INSERT,
+ "id-1",
+ "Alice",
+ 1L,
+ Date.valueOf("2026-01-01"));
+ Row updateAfter = Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "id-1",
+ "Alice updated",
+ 2L,
+ Date.valueOf("2026-01-01"));
+ Row delete = Row.ofKind(
+ RowKind.DELETE,
+ "id-1",
+ "Alice updated",
+ 3L,
+ Date.valueOf("2026-01-01"));
+ DataStream<Row> stream = context.environment.fromCollection(
+ List.of(insert, updateAfter, delete),
+ Types.ROW_NAMED(
+ new String[] {"id", "name", "event_version", "partition_date"},
+ Types.STRING,
+ Types.STRING,
+ Types.LONG,
+ Types.SQL_DATE));
+ Schema schema = Schema.newBuilder()
+ .column("id", DataTypes.STRING().notNull())
+ .column("name", DataTypes.STRING())
+ .column("event_version", DataTypes.BIGINT().notNull())
+ .column("partition_date", DataTypes.DATE().notNull())
+ .primaryKey("id")
+ .build();
+ ChangelogMode upsertMode = ChangelogMode.newBuilder()
+ .addContainedKind(RowKind.INSERT)
+ .addContainedKind(RowKind.UPDATE_AFTER)
+ .addContainedKind(RowKind.DELETE)
+ .build();
+ Table source = context.tableEnv.fromChangelogStream(stream, schema,
upsertMode);
+ context.tableEnv.createTemporaryView("orders_mutable_source", source);
+ }
+
private static void registerInsertOnlyView(
StreamTableEnvironment tableEnv,
String viewName,
diff --git a/hudi-agent-gateway/skills/hudi-architect/README.md
b/hudi-agent-gateway/skills/hudi-architect/README.md
index cb9be41ae622..e4796e0818bc 100644
--- a/hudi-agent-gateway/skills/hudi-architect/README.md
+++ b/hudi-agent-gateway/skills/hudi-architect/README.md
@@ -35,10 +35,11 @@ Target Hudi version: **1.2.0**.
- **Spark and HoodieStreamer** retain the existing full design flow,
configuration bundle, and
runnable output.
-- **Flink** combines the routing and safety foundation with one bounded
executable Flink SQL sink
- path for Hudi 1.2.0 and Flink 1.20 (1.20.1 fixtures). New-table,
single-writer, append-only COW
- designs without an external catalog can reach `CONFIG_VALIDATED` after
schema, identity, source
- changelog, append-mode, and checkpoint validation. Other Flink designs
remain fail-closed.
+- **Flink** combines the routing and safety foundation with bounded
append-only and first mutable
+ COW SQL sink paths for Hudi 1.2.0 and Flink 1.20 (1.20.1 fixtures). Mutable
eligibility requires
+ stable identity, event-time ordering, normalized UPSERT changelog, full-row
deletes when present, immutable
+ partitions, and the fixed global FLINK_STATE/bootstrap/no-TTL contract.
Other Flink designs
+ remain fail-closed.
## How to invoke
@@ -120,6 +121,9 @@ python3
hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py \
The validator emits SQL only after all safety and load-bearing checks pass.
The exact SQL is also
planned in CI against the released Hudi 1.2.0 Flink bundle and Flink 1.20.1.
+The mutable schema-2 fixture is
+`hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr3/mutable_cow.json`.
+
Credential-bearing evidence can be sanitized without executing or parsing it
as configuration:
```bash
@@ -140,9 +144,10 @@ This is **Milestone 1 of a longer arc** — meant to be
shareable and playable,
## What's out of scope in Milestone 1
-- Flink upsert/delete, MOR, existing-table composition, external catalogs,
multiple writers,
+- Flink MOR, existing-table takeover, external catalogs, multiple writers,
alternative indexes,
+ custom mergers, retract normalization, partition movement, key-only deletes,
full CDC history,
DataStream API, `HoodieFlinkStreamer`, source connector generation, and
submit/deploy commands.
- Those paths do not reuse the bounded append-only SQL output as if it were
generally applicable.
+ Those paths do not reuse the bounded SQL output as if it were generally
applicable.
- Multi-writer contention tuning — the mode, lock provider, and config bundle
are derived, but retry/timeout tuning and early conflict detection stay at
defaults (they depend on observed behavior, not design-time facts).
- Benchmarking / scale-characterization — different flow shape, future
revision.
diff --git a/hudi-agent-gateway/skills/hudi-architect/RUNBOOK.md
b/hudi-agent-gateway/skills/hudi-architect/RUNBOOK.md
index 44f15739f6f7..5ab8c8d4d05b 100644
--- a/hudi-agent-gateway/skills/hudi-architect/RUNBOOK.md
+++ b/hudi-agent-gateway/skills/hudi-architect/RUNBOOK.md
@@ -25,7 +25,7 @@ A practical guide to running the Hudi Architect agent: what
it does, how to set
(ADR) with tradeoff tables and measurable revisit conditions, a ready-to-use
`hoodie.*` config
bundle, and a runnable submit command — all pinned to **Hudi 1.2.0**. The
Flink route is pinned to
**Hudi 1.2.0 / Flink 1.20** and remains fail-closed by default. It can emit
validated Flink SQL for
-one bounded new-table, single-writer, no-external-catalog, append-only COW
sink path.
+bounded append-only and first mutable new-table, single-writer,
no-external-catalog COW paths.
---
@@ -141,8 +141,10 @@ For a Flink request, the session always includes an
ADR-shaped safety assessment
Hudi 1.2.0 source revision plus Flink 1.20.1 fixture version. Any unresolved
gate returns
`INCOMPLETE`, `BLOCKED`, or `REVIEW_REQUIRED` without SQL. A design in the
bounded executable
surface reaches `CONFIG_VALIDATED` only after the physical target and source
schemas, identity,
-`INSERT_ONLY` source changelog, COW insert append mode
(`write.insert.cluster=false`), and a
-checkpoint interval of at least 1000 ms pass deterministic validation. Its
artifacts are runtime `SET`
+source changelog, fixed write mode, and a checkpoint interval of at least 1000
ms pass deterministic
+validation. Mutable input additionally requires a stable non-null key,
event-time ordering,
+normalized UPSERT changelog, full-row deletes when present, immutable
partitions, and the fixed global
+FLINK_STATE/bootstrap/no-TTL contract. Its artifacts are runtime `SET`
statements, Hudi `CREATE TABLE`, and an explicit-column `INSERT INTO`;
deployment remains manual.
Then: land a first commit in a staging path, run your real read patterns
against it, and check the ADR's operational playbook section for what to
monitor from day one (commit duration, pending compactions, active timeline
size, small-file ratio).
@@ -151,10 +153,11 @@ Then: land a first commit in a staging path, run your
real read patterns against
Be aware of what the agent will *decline* to decide — it defers honestly
rather than guessing, but plan for these yourself:
-- **Flink executable coverage** — only a new-table, single-writer,
no-external-catalog,
- append-only COW streaming SQL sink is implemented. Mutable/upsert/delete,
MOR, existing tables,
- catalogs, multiple writers, DataStream API, `HoodieFlinkStreamer`, source
connector generation,
- and submit/deploy commands remain out of scope.
+- **Flink executable coverage** — bounded append-only and first mutable
new-table, single-writer,
+ no-external-catalog COW streaming SQL sinks are implemented. MOR,
existing-table takeover,
+ catalogs, multiple writers, alternative indexes, custom mergers, retract
normalization,
+ partition movement, key-only deletes, full CDC history, DataStream API,
`HoodieFlinkStreamer`,
+ source connector generation, and submit/deploy commands remain out of scope.
- **Multi-writer contention tuning** — on the Spark route, the agent asks
whether anything else writes the table (every tier but the lightest), derives
the concurrency mode, picks a lock provider, and emits a complete, runnable
block. What it does *not* do is tune for observed contention: lock
retry/timeout values, conflict-retry counts, and early conflict detection stay
at their defaults, because the right values depend on measured behavior rather
than design-time facts. That is Operati [...]
- **Catalog / metastore sync beyond the common paths** — on the Spark route,
the agent asks which engines query the table and derives sync config for Hive
Metastore, AWS Glue, BigQuery, and DataHub, including the partition-extractor
and MOR `_ro`/`_rt` consequences. The current Flink path detects an
external-catalog requirement but emits no catalog configuration. Not covered:
Polaris beyond pointing at the Spark catalog config,
Snowflake/Redshift-specific setup, and per-catalog auth (Ker [...]
diff --git a/hudi-agent-gateway/skills/hudi-architect/SKILL.md
b/hudi-agent-gateway/skills/hudi-architect/SKILL.md
index 0c51f03f4b37..bae7f9a110df 100644
--- a/hudi-agent-gateway/skills/hudi-architect/SKILL.md
+++ b/hudi-agent-gateway/skills/hudi-architect/SKILL.md
@@ -75,11 +75,14 @@ that manifest with options discovered from the enclosing
checkout. If the user e
another Hudi or Flink version, do not silently reuse the baseline: add a
`REVIEW_REQUIRED` finding
unless a separate verified capability reference exists.
-The Flink implementation includes the PR1 safety foundation and the first PR2
executable path. It
-may generate Flink SQL only for a new-table, confirmed-single-writer,
append-only COW streaming
-sink whose source contract is `INSERT_ONLY`. Every safety gate must pass, and
-`validate_flink_design.py` must return `CONFIG_VALIDATED`. Mutable COW, MOR,
existing-table,
-catalog, and multi-writer requests still fail closed for their later PRs.
+The Flink implementation includes the PR1 safety foundation, the PR2
append-only path, and the
+first bounded PR3 mutable COW path. Both require a new table, confirmed single
writer, no external
+catalog, streaming checkpointing, and `CONFIG_VALIDATED` from
`validate_flink_design.py`. The
+mutable path additionally requires stable non-null identity, one event-time
ordering field,
+normalized UPSERT changelog, full-row deletes when present, immutable
partitions, and the fixed global
+FLINK_STATE/bootstrap/no-TTL settings. MOR, existing-table takeover, catalog,
multi-writer,
+alternative indexes, custom mergers, retract normalization, and full CDC
history still fail
+closed.
## Flow structure
@@ -88,7 +91,7 @@ The conversation has three parts:
1. **Tier gate** — a single scoping question to figure out which downstream
questions fire.
2. **Rounds 1-3** — workload questions, gated conditionally by tier.
3. **Output** — engine-specific output: the established Spark artifacts, a
non-executable Flink
- assessment, or the bounded validator-rendered PR2 SQL artifacts.
+ assessment, or bounded validator-rendered PR2/PR3 SQL artifacts.
Load `references/question-flow.md` for the full round-by-round question list
with conditional gating.
@@ -141,7 +144,7 @@ Internal labels for these four tiers: `EXPLORATION`,
`PROTOTYPING`, `PRODUCTIONI
**What fires per tier:**
- **EXPLORATION** — Round 1 abbreviated, concept-explanation focused. May not
produce a full ADR — often a "here's what your workload would look like as a
Hudi table" narrative. Replace hard questions with explanations ("Hudi supports
Spark and Flink — Spark is most common; I'll assume Spark unless you say
otherwise").
-- **PROTOTYPING** — Round 1, then a **disclosed-defaults consent block** for
table size / partitioning / retention, then **hard-ask the non-defaultable
facts**: record key and ordering field when mutable, and whether anything else
writes the table. On an implemented executable route, the goal is a genuinely
runnable first table, not a sketch, and a prototyping ADR must not ship
placeholder values. The Flink PR2 path additionally requires a concrete target,
source contract, and a checkpoi [...]
+- **PROTOTYPING** — Round 1, then a **disclosed-defaults consent block** for
table size / partitioning / retention, then **hard-ask the non-defaultable
facts**: record key and ordering field when mutable, and whether anything else
writes the table. On an implemented executable route, the goal is a genuinely
runnable first table, not a sketch, and a prototyping ADR must not ship
placeholder values. The Flink executable paths additionally require a concrete
target, source contract, and a c [...]
- **PRODUCTIONIZING_INITIAL** — Rounds 1 + 2. Full
mutation/identity/partitioning questions. Production-safe defaults.
- **PRODUCTION_AT_SCALE** — All rounds. Full rubric. Guardrails strict. All
revisit conditions surfaced.
@@ -211,7 +214,7 @@ The Flink status vocabulary is:
- `BLOCKED` — the requested path is known to be unsupported or outside the
currently implemented
Flink capability.
- `REVIEW_REQUIRED` — a compatibility or operational risk needs human
confirmation.
-- `CONFIG_VALIDATED` — every safety gate and load-bearing PR2 design value
passed pinned static
+- `CONFIG_VALIDATED` — every safety gate and load-bearing executable design
value passed pinned static
validation, and canonical SQL was emitted. It does not prove the live
environment.
**Revisit conditions must be measurable.** Not "revisit if write amp becomes
an issue." Yes: "if p95 commit duration exceeds the ingestion interval on a COW
table above 1TB, evaluate switching to MOR — note this requires a table
rewrite, so decide before the table grows further."
diff --git
a/hudi-agent-gateway/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
b/hudi-agent-gateway/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
index b53f5dc78fb6..7aead7775916 100644
---
a/hudi-agent-gateway/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
+++
b/hudi-agent-gateway/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.md
@@ -82,7 +82,15 @@ the current checkout contains it.
| Append-only record-key posture detection | Supported |
| Auto-generated-key durability warning | Supported |
| Replay-idempotence classification | Supported |
-| Mutable COW, upsert, and deletes | `BLOCKED`; deferred to PR3 |
+| First bounded mutable COW upsert/delete path | Supported after schema-2
validation |
+| Stable non-null mutable record key | Required; auto-key is `BLOCKED` |
+| Mutable ordering | One non-null `BIGINT` or `TIMESTAMP(p<=6)` event-time
field |
+| Mutable source changelog | Normalized UPSERT (`I`, `UA`, `D`, no `UB`) |
+| Mutable delete payload | Full projected row required when deletes are
emitted |
+| Partition changes for an existing key | `BLOCKED`; immutable partition
values required |
+| Mutable index | Global `FLINK_STATE`, TTL 0, bootstrap enabled |
+| Alternative indexes, expiring state, custom mergers, or retract
normalization | `BLOCKED`; deferred |
+| Full changelog history / Hudi CDC query | Not claimed; output is
latest-state behavior |
| MOR and compaction ownership | Deferred to PR4 |
| New-table, single-writer, append-only COW SQL | Supported after static
validation |
| Stable-key DDL | `PRIMARY KEY (...) NOT ENFORCED` |
@@ -160,13 +168,33 @@ path deliberately applies a stricter 1000 ms safety floor
and blocks lower value
silently increasing them. The pinned runtime test preserves the 9/10 ms
dependency boundary, and
the Python validator tests preserve the 999/1000 ms Architect boundary.
+## Implemented PR3 mutable COW contract
+
+The schema-2 path is deliberately narrower than the full Hudi upsert surface.
It accepts only a
+new table, one writer, no external catalog, streaming checkpoints, stable
non-null identity, one
+event-time ordering field, normalized UPSERT input, immutable partitions, and
full-row deletes when
+present. All three resolved replay answers are eligible; only an unresolved
replay posture blocks
+validation. The exact ordering-field value `no_precombine` is rejected because
Hudi 1.2.0 resolves
+that sentinel to no ordering field even if the physical schema contains a
same-named column.
+The renderer fixes `COPY_ON_WRITE`, `upsert`, `EVENT_TIME_ORDERING`, global
`FLINK_STATE`, state
+TTL `0`, index bootstrap, and `changelog.enabled=false`; it accepts no
pass-through override.
+
+The pinned Flink planner fixture consumes the exact SQL golden with an
`I/UA/D` source and proves
+that the generated graph includes changelog normalization, the Hudi stream
writer, and
+`index_bootstrap`. Direct pinned-option fixtures preserve the ordering, index,
TTL, bootstrap, and
+changelog values. Python tests cover the successful golden and fail-closed
deviations.
+
+Bootstrap closes the cold-start gap for a table created and operated under
this contract: state
+loss must not make existing keys look new. It does not expand the path into
arbitrary
+existing-table takeover; that still needs the separate PR5 evidence and
compatibility flow.
+
## Status vocabulary
- `INCOMPLETE` — a required workload fact, schema, or lifecycle fact is
missing.
- `BLOCKED` — the requested path is known to be outside the currently
implemented Flink scope.
- `REVIEW_REQUIRED` — compatibility, writer topology, catalog behavior, or
another operational
risk requires human confirmation.
-- `CONFIG_VALIDATED` — every PR1 gate passed and all load-bearing PR2 design
values passed the
+- `CONFIG_VALIDATED` — every applicable gate and all load-bearing PR2 or PR3
design values passed the
pinned static validator. Canonical SQL was emitted without unresolved
placeholders.
No status claims that storage permissions, JAR deployment, source
availability, catalog
diff --git
a/hudi-agent-gateway/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
b/hudi-agent-gateway/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
index 86f60969719a..d193a36d97b6 100644
---
a/hudi-agent-gateway/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
+++
b/hudi-agent-gateway/skills/hudi-architect/references/flink-1.20-hudi-1.2.0-capabilities.toml
@@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-schema_version = 2
+schema_version = 3
baseline_id = "hudi-1.2.0-flink-1.20"
validation_surface = "initial-flink-sql-sink"
@@ -43,7 +43,26 @@ require_checkpointing = true
allow_record_key_modes = ["stable_key", "auto_key"]
allow_replay_behaviors = ["cannot_occur", "duplicates_acceptable"]
-# PR2 intentionally validates a bounded scalar physical-schema surface.
Complex,
+[mutable_executable_contract]
+contract_schema = 2
+table_lifecycle = "new"
+writer_model = "single_writer"
+external_catalog = false
+mutability = "mutable"
+source_changelog_mode = "UPSERT"
+execution_mode = "STREAMING"
+require_checkpointing = true
+record_key_mode = "stable_key"
+allow_replay_behaviors = ["cannot_occur", "duplicates_acceptable",
"must_collapse"]
+ordering_mode = "event_time"
+ordering_field_types = ["BIGINT", "TIMESTAMP"]
+reserved_ordering_field_values = ["no_precombine"]
+allow_emits_deletes = [false, true]
+delete_payload_when_deletes = "FULL_ROW"
+delete_payload_without_deletes = "NOT_APPLICABLE"
+partition_fields_mutable = false
+
+# The executable paths intentionally validate a bounded scalar physical-schema
surface. Complex,
# computed, metadata, and watermark columns remain unverified rather than
being guessed.
[physical_types]
simple = [
@@ -123,7 +142,7 @@ purpose = "Writer-side prepending of Hudi metadata fields
to the Flink row type"
id = "options_resolver"
path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java"
sha256 = "1a63ef88bbdf84cbcebb92f402557c21b6f84bdd0112726792ce22da635ad766"
-purpose = "Effective append-mode predicate"
+purpose = "Effective append-mode predicate and no-precombine ordering sentinel
resolution"
[[source_anchors]]
id = "table_factory"
@@ -149,8 +168,56 @@ path =
"hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/Tes
sha256 = "ef0ae3752cc8ed63f9b6ae83a4aa34d61b30bc163282a658a2d52135154ff15a"
purpose = "Release behavior exercised by the existing factory fixtures"
-# This is an allowlist for the first executable sink path, not a claim that it
+[[source_anchors]]
+id = "table_sink"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java"
+sha256 = "59c36ec6c97524386c4ce0fb7c70960816a6cfc9dee89e82de86f401f28e1087"
+purpose = "Sink changelog-mode negotiation for upsert writes"
+
+[[source_anchors]]
+id = "changelog_modes"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ChangelogModes.java"
+sha256 = "ca48d1fbc3cd7e1f6bd7009f8c1cf93d130284cba40a47b245e9002f1733797a"
+purpose = "Exact INSERT, UPDATE_AFTER, and DELETE upsert changelog contract"
+
+[[source_anchors]]
+id = "index_backend_factory"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/IndexBackendFactory.java"
+sha256 = "57cc58e85475ae32c35da57e9d5df7e8ac040edfd26cccb8859407a25233b722"
+purpose = "FLINK_STATE backend and zero-TTL state behavior"
+
+[[source_anchors]]
+id = "pipelines"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/Pipelines.java"
+sha256 = "9278096f0894a0ad34be09206ee06d317fda4185101833018a314e87ade5e262"
+purpose = "Index bootstrap insertion into the Flink write pipeline"
+
+[[source_anchors]]
+id = "bootstrap_operator"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperator.java"
+sha256 = "67ea9f1dede0cfe81252613bf158d5fa64dec8e3040f56d50aa6795d7b501e1d"
+purpose = "Existing key and file-location recovery into Flink state"
+
+[[source_anchors]]
+id = "bucket_assign_function"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java"
+sha256 = "89c36a6c64fb4524505b0ee4c1ab7de68b4a2b9343df659bbc873ecc3b259cad"
+purpose = "Global-index record routing and partition-change behavior"
+
+[[source_anchors]]
+id = "streamer_util"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java"
+sha256 = "c771e137ebcd45c94a941857a246dc741e1dde2c0e0158900fd48fe8884f83e2"
+purpose = "Write-side record merge mode inference and table-property
persistence"
+
+# This is an allowlist for the bounded executable sink paths, not a claim that
it
# enumerates every option available in Hudi 1.2.0.
+[[verified_options]]
+key = "changelog.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options", "table_sink", "changelog_modes"]
+source_symbol = "CHANGELOG_ENABLED"
+
[[verified_options]]
key = "hoodie.datasource.write.partitionpath.field"
value_type = "string"
@@ -163,10 +230,40 @@ value_type = "string"
source_anchors = ["flink_options", "key_generator_options"]
source_symbol = "RECORD_KEY_FIELD"
+[[verified_options]]
+key = "hoodie.write.record.merge.mode"
+value_type = "string"
+source_anchors = ["flink_options", "streamer_util"]
+source_symbol = "RECORD_MERGE_MODE"
+
+[[verified_options]]
+key = "index.bootstrap.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options", "pipelines", "bootstrap_operator"]
+source_symbol = "INDEX_BOOTSTRAP_ENABLED"
+
+[[verified_options]]
+key = "index.global.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options", "bucket_assign_function"]
+source_symbol = "INDEX_GLOBAL_ENABLED"
+
+[[verified_options]]
+key = "index.state.ttl"
+value_type = "double"
+source_anchors = ["flink_options", "index_backend_factory"]
+source_symbol = "INDEX_STATE_TTL"
+
+[[verified_options]]
+key = "index.type"
+value_type = "string"
+source_anchors = ["flink_options", "index_backend_factory"]
+source_symbol = "INDEX_TYPE"
+
[[verified_options]]
key = "ordering.fields"
value_type = "string"
-source_anchors = ["flink_options"]
+source_anchors = ["flink_options", "options_resolver"]
source_symbol = "ORDERING_FIELDS"
[[verified_options]]
@@ -208,6 +305,46 @@ key = "write.operation"
value = "insert"
applies_to = "pr2-append-only-cow"
+[[required_effective_settings]]
+key = "changelog.enabled"
+value = false
+applies_to = "pr3-mutable-cow"
+
+[[required_effective_settings]]
+key = "hoodie.write.record.merge.mode"
+value = "EVENT_TIME_ORDERING"
+applies_to = "pr3-mutable-cow"
+
+[[required_effective_settings]]
+key = "index.bootstrap.enabled"
+value = true
+applies_to = "pr3-mutable-cow"
+
+[[required_effective_settings]]
+key = "index.global.enabled"
+value = true
+applies_to = "pr3-mutable-cow"
+
+[[required_effective_settings]]
+key = "index.state.ttl"
+value = 0.0
+applies_to = "pr3-mutable-cow"
+
+[[required_effective_settings]]
+key = "index.type"
+value = "FLINK_STATE"
+applies_to = "pr3-mutable-cow"
+
+[[required_effective_settings]]
+key = "table.type"
+value = "COPY_ON_WRITE"
+applies_to = "pr3-mutable-cow"
+
+[[required_effective_settings]]
+key = "write.operation"
+value = "upsert"
+applies_to = "pr3-mutable-cow"
+
[[implemented_acceptance_checks]]
id = "FLINK_BINARY_ROUTING_FIELD_UNSUPPORTED"
requirement = "Reject binary record-key and partition fields whose Hudi 1.2.0
routing strings depend on Java object identity."
@@ -255,3 +392,27 @@ id = "FLINK_CHECKPOINT_INTERVAL_UNSUPPORTED"
requirement = "Reject checkpoint intervals below the 1000 ms Architect safety
floor while preserving the pinned Flink 10 ms runtime boundary as evidence."
python_test = "test_pr2_enforces_checkpoint_interval_safety_floor"
java_test = "testPinnedRuntimeCheckpointIntervalBoundary"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED"
+requirement = "Require a normalized UPSERT source and verify the Hudi sink
consumes INSERT, UPDATE_AFTER, and DELETE without UPDATE_BEFORE."
+python_test = "test_pr3_rejects_unsupported_source_changelog"
+java_test = "testMutableCowSinkAndUpsertPlan"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MUTABLE_ORDERING_FIELD_INVALID"
+requirement = "Require one non-null BIGINT or TIMESTAMP(p<=6) event-time
ordering field and reject Hudi's no_precombine sentinel."
+python_test = "test_pr3_rejects_invalid_ordering_fields"
+java_test = "testPinnedMutableOrderingConfiguration"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED"
+requirement = "Pin global FLINK_STATE with no state expiry for replay-safe
mutable routing."
+python_test = "test_pr3_rejects_unsafe_index_configuration"
+java_test = "testPinnedMutableIndexConfiguration"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED"
+requirement = "Require index bootstrap so a cold restart does not treat
existing keys as new."
+python_test = "test_pr3_requires_index_bootstrap"
+java_test = "testPinnedMutableIndexBootstrapConfiguration"
diff --git
a/hudi-agent-gateway/skills/hudi-architect/references/flink-config-templates.md
b/hudi-agent-gateway/skills/hudi-architect/references/flink-config-templates.md
index 2796fa902e44..2759417f0934 100644
---
a/hudi-agent-gateway/skills/hudi-architect/references/flink-config-templates.md
+++
b/hudi-agent-gateway/skills/hudi-architect/references/flink-config-templates.md
@@ -15,7 +15,7 @@ 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.
-->
-# Flink output template — first executable PR2 sink
+# Flink output template — bounded executable sinks
Use this file only after the Flink route is selected. The PR1 safety gates
still control entry to
the executable path. Never fall back to shared Spark configuration templates.
@@ -30,8 +30,8 @@ Verified baseline: Apache Hudi 1.2.0 / Apache Flink 1.20
Baseline ID: hudi-1.2.0-flink-1.20
Hudi source revision: f05c83f2b97732de7a558ff9b26959e1139c05f5
Fixture version: Apache Flink 1.20.1
-Capability manifest schema: 2
-Design contract schema: 1
+Capability manifest schema: 3
+Design contract schema: <1 append-only | 2 mutable COW>
Confirmed facts:
- <facts established by explicit answers>
@@ -65,26 +65,28 @@ Verified baseline: Apache Hudi 1.2.0 / Apache Flink 1.20
Baseline ID: hudi-1.2.0-flink-1.20
Hudi source revision: f05c83f2b97732de7a558ff9b26959e1139c05f5
Fixture version: Apache Flink 1.20.1
-Capability manifest schema: 2
-Design contract schema: 1
+Capability manifest schema: 3
+Design contract schema: <1 append-only | 2 mutable COW>
Confirmed facts:
- New table; confirmed single writer; no current external-catalog requirement.
-- Append-only input; replay behavior: <cannot occur | duplicates accepted>.
-- Identity: <stable key fields | explicitly accepted auto-generated keys>.
+- Workload: <append-only or mutable, with confirmed replay behavior>.
+- Identity: <stable key fields | explicitly accepted append-only
auto-generated keys>.
+- Mutable only: one event-time ordering field; immutable partition values;
full-row delete posture.
Source contract:
- Existing table: <source identifier>
- Expected physical schema: <Avro-compatible field names, bounded types,
nullability>
-- Changelog: INSERT_ONLY
+- Changelog: <INSERT_ONLY for schema 1 | normalized UPSERT (I/UA/D, no UB) for
schema 2>
- The source connector and live source availability were not validated.
Target contract:
- Table: <target identifier>
- Path: <credential-free concrete path>
- Table type: COPY_ON_WRITE
-- Operation: insert
-- Insert clustering: false
+- Operation: <insert for schema 1 | upsert for schema 2>
+- Path-specific fixed settings: <insert clustering false | EVENT_TIME_ORDERING
plus global
+ FLINK_STATE, TTL 0, bootstrap true, and changelog false>
Runtime contract:
- Execution: streaming
@@ -114,8 +116,7 @@ SET 'execution.checkpointing.interval' = '<milliseconds,
minimum 1000> ms';
Flink 1.20.1 has a lower runtime boundary of 10 ms, but the bounded Architect
path intentionally
requires at least 1000 ms and never raises a supplied value implicitly.
-The generated `CREATE TABLE` contains a Hudi `WITH` block with exactly the
bounded load-bearing
-settings:
+For schema 1, the generated `CREATE TABLE` contains exactly these bounded
load-bearing settings:
```sql
WITH (
@@ -136,22 +137,48 @@ even a verified option when the bounded renderer has no
canonical place for it.
The angle-bracket values in this reference explain the shape only. They are
forbidden in an actual
`CONFIG_VALIDATED` artifact; the validator requires concrete values.
+For schema 2 mutable COW, the canonical block is:
+
+```sql
+WITH (
+ 'connector' = 'hudi',
+ 'path' = '<concrete credential-free target path>',
+ 'table.type' = 'COPY_ON_WRITE',
+ 'write.operation' = 'upsert',
+ 'ordering.fields' = '<one non-null BIGINT or TIMESTAMP(p<=6) field>',
+ 'hoodie.write.record.merge.mode' = 'EVENT_TIME_ORDERING',
+ 'index.type' = 'FLINK_STATE',
+ 'index.global.enabled' = 'true',
+ 'index.state.ttl' = '0',
+ 'index.bootstrap.enabled' = 'true',
+ 'changelog.enabled' = 'false'
+)
+```
+
+The mutable source must declare the same primary key as the target and
normalized UPSERT
+changelog (`I`, `UA`, `D`, no `UB`). Deletes, when emitted, must be full-row;
partition values are immutable.
+This is a latest-state sink contract; it is not full changelog-history or
CDC-query support.
+
## Deployment checks
Always retain these outside the static success claim:
- Deploy `org.apache.hudi:hudi-flink1.20-bundle:1.2.0` with the target Flink
1.20 environment.
- Verify storage permissions and the concrete target path.
-- Verify that the declared source table exists and preserves the expected
insert-only contract.
+- Verify that the declared source table exists and preserves its validated
INSERT_ONLY or
+ normalized UPSERT contract.
- Configure durable checkpoint storage and a restart strategy appropriate to
the environment.
- Confirm the single-writer inventory again immediately before deployment.
- After recovery, verify that completed checkpoints correspond to completed
Hudi instants.
-## Still forbidden in PR2
+## Still forbidden in the bounded paths
- A complete Kafka, CDC, JDBC, or other source connector job.
- A Flink deployment or submit command.
- A Spark fallback.
-- OCC, NBCC, lock-provider, index, catalog, MOR, compaction, or clustering
configuration.
+- OCC, NBCC, lock-provider, alternative index, catalog, MOR, compaction, or
clustering
+ configuration.
+- Existing-table takeover, multi-writer operation, expiring state index,
key-only deletes,
+ retract-stream normalization, partition movement, custom mergers, or full
CDC history.
- A claim that static validation checked live storage, the deployed JAR, the
live source, active
writers, checkpoint completion, or end-to-end exactly-once behavior.
diff --git
a/hudi-agent-gateway/skills/hudi-architect/references/flink-decision-overrides.md
b/hudi-agent-gateway/skills/hudi-architect/references/flink-decision-overrides.md
index 334f99c97253..92dd6baae6cb 100644
---
a/hudi-agent-gateway/skills/hudi-architect/references/flink-decision-overrides.md
+++
b/hudi-agent-gateway/skills/hudi-architect/references/flink-decision-overrides.md
@@ -15,7 +15,7 @@ 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.
-->
-# Flink decision overrides — first executable PR2 path
+# Flink decision overrides — bounded executable SQL paths
These rules constrain the shared Hudi Architect decisions after the Flink
route is selected. They
do not replace the shared table-design rules and do not form a standalone
planner API.
@@ -45,8 +45,9 @@ do not replace the shared table-design rules and do not form
a standalone planne
safety floor.
- Do not equate a stable record key with replay idempotence.
- Do not reuse this baseline for another Hudi or Flink version.
-- Emit executable output only when `validate_flink_design.py` returns
`CONFIG_VALIDATED` for the
- complete version-1 design contract. Never hand-write around a validator
finding.
+- Emit executable output only when `validate_flink_design.py` returns
`CONFIG_VALIDATED` for a
+ complete version-1 append-only or version-2 mutable contract. Never
hand-write around a
+ validator finding.
## Gate outcomes
@@ -62,12 +63,12 @@ do not replace the shared table-design rules and do not
form a standalone planne
| External catalog required | `FLINK_EXTERNAL_CATALOG_REVIEW` |
`REVIEW_REQUIRED` | No |
| Catalog requirement unknown | `FLINK_CATALOG_REQUIREMENT_UNRESOLVED` |
`REVIEW_REQUIRED` | No |
| Physical schema missing, insufficient, or available only through an
unreadable pointer | `FLINK_PHYSICAL_SCHEMA_REQUIRED` | `INCOMPLETE` | No |
-| Mutable workload | `FLINK_MUTABLE_COW_DEFERRED` | `BLOCKED` | No |
+| Mutable workload supplied through the version-1 append contract |
`FLINK_MUTABLE_COW_DEFERRED` | `BLOCKED` | No |
| Mutability unknown | `FLINK_MUTABILITY_REQUIRED` | `INCOMPLETE` | No |
| Record-key posture unknown | `FLINK_RECORD_KEY_POSTURE_REQUIRED` |
`INCOMPLETE` | No |
| No stable key and auto-key acceptance is pending or unanswered |
`FLINK_AUTO_KEY_ACCEPTANCE_REQUIRED` | `INCOMPLETE` | No |
| No stable key and the auto-key posture is explicitly declined |
`FLINK_AUTO_KEY_DECLINED` | `BLOCKED` | No |
-| Replay copies must collapse | `FLINK_REPLAY_IDEMPOTENCE_DEFERRED` |
`BLOCKED` | No |
+| Append-only replay copies must collapse |
`FLINK_REPLAY_IDEMPOTENCE_DEFERRED` | `BLOCKED` | No |
| Replay behavior unknown | `FLINK_REPLAY_BEHAVIOR_UNRESOLVED` |
`REVIEW_REQUIRED` | No |
| Stable or explicit record-key option references a missing field |
`FLINK_RECORD_KEY_FIELD_MISSING` | `BLOCKED` | No |
| Stable record-key field is nullable | `FLINK_RECORD_KEY_NULLABLE` |
`BLOCKED` | No |
@@ -83,13 +84,22 @@ do not replace the shared table-design rules and do not
form a standalone planne
| Source changelog is not `INSERT_ONLY` |
`FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY` | `BLOCKED` | No |
| Effective `write.insert.cluster` is not `false` |
`FLINK_APPEND_MODE_CLUSTERING_ENABLED` | `BLOCKED` | No |
| Table type, operation, or execution mode is outside PR2 |
`FLINK_PR2_WRITE_PATH_UNSUPPORTED` | `BLOCKED` | No |
+| Mutable path uses auto-generated keys | `FLINK_MUTABLE_AUTO_KEY_UNSUPPORTED`
| `BLOCKED` | No |
+| Mutable source is not normalized UPSERT (`I`, `UA`, `D`, no `UB`) |
`FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED` | `BLOCKED` | No |
+| Mutable ordering field is absent, nullable, outside BIGINT/TIMESTAMP(p≤6),
or is `no_precombine` | `FLINK_MUTABLE_ORDERING_FIELD_INVALID` | `BLOCKED` | No
|
+| Mutable ordering decision is absent or not event-time |
`FLINK_MUTABLE_ORDERING_FIELD_REQUIRED` | `INCOMPLETE` | No |
+| A mutable delete does not carry the full projected row |
`FLINK_MUTABLE_DELETE_PAYLOAD_UNSUPPORTED` | `BLOCKED` | No |
+| Partition values can change for an existing key |
`FLINK_MUTABLE_PARTITION_EVOLUTION_UNSUPPORTED` | `BLOCKED` | No |
+| Mutable index is not global `FLINK_STATE` with TTL 0 |
`FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED` | `BLOCKED` | No |
+| FLINK_STATE index bootstrap is not enabled |
`FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED` | `BLOCKED` | No |
+| Table type, operation, merge mode, changelog setting, or execution mode is
outside PR3 | `FLINK_PR3_WRITE_PATH_UNSUPPORTED` | `BLOCKED` | No |
| Checkpointing is explicitly disabled | `FLINK_CHECKPOINTING_REQUIRED` |
`BLOCKED` | No |
| Checkpoint interval is absent or non-positive |
`FLINK_CHECKPOINT_INTERVAL_REQUIRED` | `INCOMPLETE` | No |
| Positive checkpoint interval is below the 1000 ms Architect safety floor |
`FLINK_CHECKPOINT_INTERVAL_UNSUPPORTED` | `BLOCKED` | No |
| Target table, target path, source table, or another load-bearing value is a
placeholder | `FLINK_LOAD_BEARING_VALUE_REQUIRED` | `INCOMPLETE` | No |
| Design-contract structure is malformed or contains an ignored field |
`FLINK_DESIGN_CONTRACT_INVALID` | `INCOMPLETE` | No |
| A connector option is outside the pinned allowlist |
`FLINK_OPTION_NOT_VERIFIED` | `REVIEW_REQUIRED` | No |
-| Complete PR2 contract passes deterministic validation | — |
`CONFIG_VALIDATED` | Yes |
+| Complete PR2 or PR3 contract passes deterministic validation | — |
`CONFIG_VALIDATED` | Yes |
## Final-status precedence
@@ -103,7 +113,7 @@ After collection, choose one final status with this
precedence:
1. A known unsupported or not-yet-implemented path makes the result `BLOCKED`.
2. Otherwise, an unverified compatibility or operational risk makes it
`REVIEW_REQUIRED`.
3. Otherwise, a missing required fact makes it `INCOMPLETE`.
-4. Only if no finding exists, every PR1 safety gate passed, and the PR2 design
validator emitted
+4. Only if no finding exists, every applicable safety gate passed, and the
design validator emitted
canonical SQL, return `CONFIG_VALIDATED` with executable eligibility `true`.
Status precedence changes only the single summary status; it never removes a
lower-precedence
@@ -112,9 +122,14 @@ only bounded static validation against the pinned release
contract.
## Record-key and replay classification
-| Mutability | Stable business key | Replay requirement | PR2 result |
+| Mutability | Stable business key | Replay requirement | Result |
|---|---|---|---|
-| Mutable | Required | Any | `BLOCKED` — mutable COW is deferred |
+| Mutable | Yes | Replays impossible | Continue into the bounded PR3 mutable
contract |
+| Mutable | Yes | Duplicates acceptable | Continue; record duplicate tolerance
|
+| Mutable | Yes | Copies must collapse | Continue into the bounded PR3 mutable
contract |
+| Mutable | Yes | Replay behavior unknown | `REVIEW_REQUIRED` |
+| Mutable | No | Any | `BLOCKED` — auto-key cannot express mutable identity |
+| Mutable | Unknown | Any | `INCOMPLETE` |
| Append-only | Yes | Replays impossible | Continue safety gates; stable key
preferred |
| Append-only | Yes | Duplicates acceptable | Continue; record duplicate
tolerance |
| Append-only | Yes | Copies must collapse | `BLOCKED` — upsert-capable design
required |
@@ -156,8 +171,8 @@ and executable eligibility, not exact natural-language
wording.
| `F05_CATALOG` | External consumer requires catalog visibility |
`REVIEW_REQUIRED` | No |
| `F06_SCHEMA` | Physical schema is missing | `INCOMPLETE` | No |
| `F06_SCHEMA_POINTER` | Only a schema location is supplied; field names and
types are unavailable | `INCOMPLETE` | No |
-| `F07_MUTABLE` | Updates or deletes occur | `BLOCKED` | No |
-| `F08_REPLAY_COLLAPSE` | Independent replay must deduplicate | `BLOCKED` | No
|
+| `F07_MUTABLE` | Otherwise-safe mutable workload enters the schema-2
candidate flow | Proceed to PR3 validation | Not decided by gates alone |
+| `F08_REPLAY_COLLAPSE` | Append-only independent replay must deduplicate |
`BLOCKED` | No |
| `F09_REPLAY_UNKNOWN` | Replay behavior is unknown | `REVIEW_REQUIRED` | No |
| `F10_SAFE_APPEND` | Baseline new-table single-writer append-only path with a
stable key passes every gate | Proceed to PR2 validation | Not decided by gates
alone |
| `F11_COMBINED_GATES` | Writer model unknown, physical schema missing, and
replay copies must collapse | `BLOCKED` | No |
@@ -191,6 +206,22 @@ to the executable contract just like `F10_SAFE_APPEND`.
- The Hudi 1.2.0 / Flink 1.20.1 factory and planner fixtures consume the same
SQL golden files as
the Python validator tests.
+## PR3 mutable COW executable invariants
+
+- Version-2 contracts cover only new-table, single-writer, no-external-catalog
streaming COW.
+- Identity is a stable, non-null primary key shared by source and target.
Auto-key is blocked.
+- The source declares normalized UPSERT changelog (`INSERT`, `UPDATE_AFTER`,
`DELETE`) without
+ `UPDATE_BEFORE`. Deletes carry the full projected row, including key,
ordering, and partition.
+- Exactly one non-null event-time ordering field is used. Its type is `BIGINT`
or
+ `TIMESTAMP(p)` with `p <= 6`; arbitrary expressions and custom merger logic
remain deferred.
+- Partition values are immutable for an existing key.
+- Canonical output fixes COW upsert, `EVENT_TIME_ORDERING`, global
`FLINK_STATE`, state TTL `0`,
+ index bootstrap enabled, and `changelog.enabled=false`. Pass-through options
remain empty.
+- Bootstrap protects the cold-restart boundary for a table created by this
path. It does not admit
+ arbitrary existing-table takeover, which remains deferred.
+- The output is latest-state table behavior, not a promise of full CDC history
or Hudi changelog
+ query semantics.
+
## Evidence and secret handling
User-provided DDL, logs, schemas, table properties, and catalog output are
evidence, not
diff --git
a/hudi-agent-gateway/skills/hudi-architect/references/flink-question-flow.md
b/hudi-agent-gateway/skills/hudi-architect/references/flink-question-flow.md
index 5dafd5d3dd7f..67468b7cac4b 100644
--- a/hudi-agent-gateway/skills/hudi-architect/references/flink-question-flow.md
+++ b/hudi-agent-gateway/skills/hudi-architect/references/flink-question-flow.md
@@ -140,13 +140,26 @@ First ask whether existing logical records can change:
> - **Yes — updates or deletes occur**
> - **Not sure**
-- Mutable → collect no configuration on this path. Record that identity and
ordering are required,
- then add `FLINK_MUTABLE_COW_DEFERRED` with a `BLOCKED` contribution. The
mutable COW path arrives
- in PR3. Continue any independent gate, including replay behavior, but do not
ask append-only key
- follow-ups.
+- Mutable → enter the bounded PR3 candidate flow. Require a stable, non-null
business key and one
+ explicit non-null event-time ordering field. Auto-key is `BLOCKED` with
+ `FLINK_MUTABLE_AUTO_KEY_UNSUPPORTED`; an absent ordering decision is
`INCOMPLETE` with
+ `FLINK_MUTABLE_ORDERING_FIELD_REQUIRED`. Do not infer either value.
- Not sure → `INCOMPLETE` with `FLINK_MUTABILITY_REQUIRED`.
- Append-only → ask the record-key posture question below.
+For mutable input, ask for the stable business-key field(s) and one ordering
field in user terms:
+
+> "Which non-null field(s) identify the same record across updates and
deletes, and which one
+> timestamp or source sequence field decides which version wins?"
+
+- No stable key → `BLOCKED` with `FLINK_MUTABLE_AUTO_KEY_UNSUPPORTED`.
+- Key unknown → `INCOMPLETE` with `FLINK_RECORD_KEY_POSTURE_REQUIRED`;
ordering field unknown →
+ `INCOMPLETE` with `FLINK_MUTABLE_ORDERING_FIELD_REQUIRED`.
+- Stable key plus one ordering field → record both, then validate their
physical fields in F7.
+ The ordering field must be non-null `BIGINT` or `TIMESTAMP(p<=6)`; do not
infer a cast or use
+ ingestion time as a hidden default. Reject the exact value `no_precombine`:
Hudi 1.2.0 treats it
+ as a sentinel that disables ordering even when a physical column has that
name.
+
For append-only input:
> "Does the data contain stable field(s) that identify the same business
> record across retries
@@ -194,8 +207,8 @@ For append-only input:
- Cannot occur → continue and record the confirmed assumption boundary.
- Duplicates acceptable → continue and record duplicate tolerance in the ADR.
-- Copies must collapse → `BLOCKED` with `FLINK_REPLAY_IDEMPOTENCE_DEFERRED`;
this requires a
- stable key and an upsert-capable path.
+- Copies must collapse → continue only for an otherwise eligible mutable
schema-2 path. For the
+ append-only schema-1 path, use `FLINK_REPLAY_IDEMPOTENCE_DEFERRED` with
`BLOCKED`.
- Not sure → `REVIEW_REQUIRED` with `FLINK_REPLAY_BEHAVIOR_UNRESOLVED`.
## F7 — Physical table and source contract
@@ -207,8 +220,10 @@ Enter this stage only when F0-F6 have no
status-contributing finding. Collect, w
- Partition fields, or an explicit unpartitioned decision.
- Stable record-key fields when F5 found a stable business key.
- The existing source-table identifier and its expected physical columns.
+- For mutable input: source primary-key fields, exactly one ordering field,
whether deletes occur,
+ whether deletes contain the complete projected row, and whether partition
values can change.
-PR2 validates a bounded scalar Flink SQL type surface recorded in the
capability manifest.
+The executable paths validate a bounded scalar Flink SQL type surface recorded
in the capability manifest.
Nested, computed, metadata, and watermark columns remain `REVIEW_REQUIRED` with
`FLINK_SCHEMA_TYPE_UNVERIFIED` until covered by a pinned fixture. A field
referenced by the
record key or partition list must exist. Stable-key columns must be `NOT
NULL`. Every physical
@@ -251,9 +266,11 @@ explicit projection; never use `SELECT *`, infer casts, or
invent a source DDL.
> - **Updates or deletes can occur**
> - **Not sure**
-- Inserts only → record `INSERT_ONLY` and continue.
-- Updates or deletes → add `FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY` with a
`BLOCKED`
- contribution. Mutable change semantics arrive in PR3.
+- Inserts only → record `INSERT_ONLY` and continue on the append path.
+- Updates or deletes → for a mutable candidate, require normalized UPSERT
changelog containing
+ `INSERT`, `UPDATE_AFTER`, and optional `DELETE`, with no `UPDATE_BEFORE`.
Otherwise add
+ `FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED`. For an append contract, retain
+ `FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY`.
- Not sure → add `FLINK_SOURCE_CONTRACT_REQUIRED` with an `INCOMPLETE`
contribution.
This check is independent of sink-factory construction. Hudi 1.2.0 can
construct a sink without
@@ -283,13 +300,14 @@ deployment responsibilities. If known rate and
active-partition evidence indicat
data per checkpoint, surface `FLINK_CHECKPOINT_SMALL_FILE_RISK`; do not enable
clustering or tune
file sizing in PR2.
-## PR2 validation and completion
+## PR2/PR3 validation and completion
-After all applicable questions, build a version-1 JSON design contract
containing the confirmed
-safety facts, physical schemas, identity mode, source contract, fixed write
settings, and runtime
-contract. Do not insert placeholders for the target table, path, source table,
or checkpoint
-interval. Record an explicitly empty pass-through connector-option map; PR2
never accepts an
-option that the canonical renderer would ignore.
+After all applicable questions, build a version-1 JSON contract for
append-only or a version-2
+JSON contract for mutable COW. Both contain confirmed safety facts, physical
schemas, identity,
+source, write, and runtime contracts. Version 2 additionally records
event-time ordering,
+normalized UPSERT, full-row delete posture, immutable partition fields, and
the fixed global
+FLINK_STATE/bootstrap/no-TTL contract. Do not insert placeholders. Record an
explicitly empty
+pass-through connector-option map; neither renderer accepts an ignored option.
Run:
diff --git
a/hudi-agent-gateway/skills/hudi-architect/references/flink-warnings.md
b/hudi-agent-gateway/skills/hudi-architect/references/flink-warnings.md
index 0a5d8d48e1a0..ef66e6f7bc89 100644
--- a/hudi-agent-gateway/skills/hudi-architect/references/flink-warnings.md
+++ b/hudi-agent-gateway/skills/hudi-architect/references/flink-warnings.md
@@ -120,12 +120,13 @@ ordering, and partition references can eventually be
checked.
## FLINK_MUTABLE_COW_DEFERRED
-**Trigger:** Existing logical records can be updated or deleted.
+**Trigger:** Existing logical records can be updated or deleted, but the
request is still using the
+version-1 append contract.
-**Message:** Mutable COW requires identity, ordering, and an upsert-capable
path that is deferred
-to PR3.
+**Message:** Mutable COW requires the explicit version-2 identity, ordering,
changelog, delete,
+partition, and index safety contract.
-**Action:** `BLOCKED`; collect independent facts but generate no executable
configuration.
+**Action:** `BLOCKED`; qualify the request for schema 2 or generate no
executable configuration.
## FLINK_MUTABILITY_REQUIRED
@@ -188,20 +189,21 @@ status.
## FLINK_REPLAY_IDEMPOTENCE_DEFERRED
-**Trigger:** Replayed or backfilled copies can occur and must collapse into
one table record.
+**Trigger:** On the append contract, replayed or backfilled copies can occur
and must collapse into
+one table record.
**Message:** This requires stable record identity and an upsert-capable
design. The append-only
insert path cannot be presented as deduplicating.
-**Action:** `BLOCKED` until the mutable COW path is implemented.
+**Action:** `BLOCKED`; use the mutable schema-2 contract only if all of its
gates are satisfied.
## FLINK_REPLAY_BEHAVIOR_UNRESOLVED
**Trigger:** The user cannot determine whether retries, replays, or backfills
can repeat logical
records or whether repeated copies must collapse.
-**Message:** Replay safety is unresolved, so the append-only insert path
cannot be presented as
-idempotent.
+**Message:** Replay safety is unresolved, so an executable append-only or
mutable path cannot be
+validated.
**Action:** `REVIEW_REQUIRED`; retain the uncertainty and withhold executable
output.
@@ -306,7 +308,7 @@ rename it.
**Trigger:** An explicitly projected source field is missing or differs in
type or nullability from
the target field.
-**Message:** PR2 does not infer casts, aliases, or schema reconciliation.
+**Message:** The bounded executable paths do not infer casts, aliases, or
schema reconciliation.
**Action:** `BLOCKED`; require an explicit compatible source contract.
@@ -315,7 +317,7 @@ the target field.
**Trigger:** The declared source changelog contains updates or deletes.
**Message:** Sink-factory construction does not prove the complete
source-to-sink statement is
-append-only. Mutable change semantics are deferred to PR3.
+append-only. An eligible mutable request must use the separate schema-2
contract.
**Action:** `BLOCKED`; do not generate executable append SQL.
@@ -336,6 +338,83 @@ Checking only `write.operation=insert` is insufficient.
**Action:** `BLOCKED`; route mutable COW to PR3 and MOR to PR4.
+## FLINK_MUTABLE_AUTO_KEY_UNSUPPORTED
+
+**Trigger:** A mutable contract has no stable business key.
+
+**Message:** Upsert and delete semantics require durable identity across
events and restarts.
+
+**Action:** `BLOCKED`; require a stable, non-null source and target primary
key.
+
+## FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED
+
+**Trigger:** The mutable source is not normalized UPSERT (`I`, `UA`, `D`,
without `UB`).
+
+**Message:** The first mutable sink path does not normalize retract streams or
infer changelog.
+
+**Action:** `BLOCKED`; normalize upstream or defer the path.
+
+## FLINK_MUTABLE_ORDERING_FIELD_REQUIRED
+
+**Trigger:** Event-time ordering or its single ordering field is missing.
+
+**Message:** The validator cannot select conflict-resolution semantics or
invent a field.
+
+**Action:** `INCOMPLETE`; obtain one explicit ordering field.
+
+## FLINK_MUTABLE_ORDERING_FIELD_INVALID
+
+**Trigger:** The ordering field is absent, nullable, not
BIGINT/TIMESTAMP(p<=6), or is the exact
+Hudi `no_precombine` sentinel.
+
+**Message:** The bounded event-time merger surface does not cover nullable or
other orderings, and
+`no_precombine` disables ordering instead of selecting a same-named physical
field.
+
+**Action:** `BLOCKED`; correct the source contract without casting silently.
+
+## FLINK_MUTABLE_DELETE_PAYLOAD_UNSUPPORTED
+
+**Trigger:** Deletes omit projected key, ordering, partition, or row values.
+
+**Message:** The first path accepts only full-row deletes; key-only delete
normalization is not
+implemented.
+
+**Action:** `BLOCKED`; normalize upstream or defer.
+
+## FLINK_MUTABLE_PARTITION_EVOLUTION_UNSUPPORTED
+
+**Trigger:** An existing key can change its partition value.
+
+**Message:** Cross-partition movement is deliberately outside this first
mutable path.
+
+**Action:** `BLOCKED`; use immutable partition fields or an unpartitioned
design.
+
+## FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED
+
+**Trigger:** Index type, scope, or TTL differs from global FLINK_STATE with
TTL 0.
+
+**Message:** Alternative indexes and expiring state need separate correctness
evidence.
+
+**Action:** `BLOCKED`; do not weaken the canonical index contract.
+
+## FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED
+
+**Trigger:** Index bootstrap is not explicitly enabled.
+
+**Message:** A cold start with empty state could otherwise miss keys already
present in the table.
+
+**Action:** `BLOCKED`; enable bootstrap. This does not authorize arbitrary
existing-table takeover.
+
+## FLINK_PR3_WRITE_PATH_UNSUPPORTED
+
+**Trigger:** Mutable table type, operation, merge mode, changelog setting, or
runtime leaves the
+bounded COW upsert path.
+
+**Message:** PR3 implements one conservative latest-state mutable route, not
the complete Hudi
+mutable configuration surface.
+
+**Action:** `BLOCKED`; retain the request as a later capability rather than
guessing settings.
+
## FLINK_CHECKPOINTING_REQUIRED
**Trigger:** Checkpointing is explicitly disabled for the streaming sink.
diff --git
a/hudi-agent-gateway/skills/hudi-architect/validate_flink_capabilities.py
b/hudi-agent-gateway/skills/hudi-architect/validate_flink_capabilities.py
index 3c8498190769..db9cf70cbc32 100644
--- a/hudi-agent-gateway/skills/hudi-architect/validate_flink_capabilities.py
+++ b/hudi-agent-gateway/skills/hudi-architect/validate_flink_capabilities.py
@@ -40,7 +40,7 @@ DEFAULT_MANIFEST = (
SKILL_DIR / "references" / "flink-1.20-hudi-1.2.0-capabilities.toml"
)
-EXPECTED_SCHEMA_VERSION = 2
+EXPECTED_SCHEMA_VERSION = 3
EXPECTED_BASELINE_ID = "hudi-1.2.0-flink-1.20"
EXPECTED_HUDI_VERSION = "1.2.0"
EXPECTED_HUDI_SOURCE_REVISION = "f05c83f2b97732de7a558ff9b26959e1139c05f5"
@@ -62,6 +62,46 @@ EXPECTED_REQUIRED_SETTINGS = [
"value": "insert",
"applies_to": "pr2-append-only-cow",
},
+ {
+ "key": "changelog.enabled",
+ "value": False,
+ "applies_to": "pr3-mutable-cow",
+ },
+ {
+ "key": "hoodie.write.record.merge.mode",
+ "value": "EVENT_TIME_ORDERING",
+ "applies_to": "pr3-mutable-cow",
+ },
+ {
+ "key": "index.bootstrap.enabled",
+ "value": True,
+ "applies_to": "pr3-mutable-cow",
+ },
+ {
+ "key": "index.global.enabled",
+ "value": True,
+ "applies_to": "pr3-mutable-cow",
+ },
+ {
+ "key": "index.state.ttl",
+ "value": 0.0,
+ "applies_to": "pr3-mutable-cow",
+ },
+ {
+ "key": "index.type",
+ "value": "FLINK_STATE",
+ "applies_to": "pr3-mutable-cow",
+ },
+ {
+ "key": "table.type",
+ "value": "COPY_ON_WRITE",
+ "applies_to": "pr3-mutable-cow",
+ },
+ {
+ "key": "write.operation",
+ "value": "upsert",
+ "applies_to": "pr3-mutable-cow",
+ },
]
EXPECTED_PHYSICAL_TYPES = {
"simple": [
@@ -140,6 +180,22 @@ EXPECTED_ACCEPTANCE_EVIDENCE = {
"test_pr2_enforces_checkpoint_interval_safety_floor",
"testPinnedRuntimeCheckpointIntervalBoundary",
),
+ "FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED": (
+ "test_pr3_rejects_unsupported_source_changelog",
+ "testMutableCowSinkAndUpsertPlan",
+ ),
+ "FLINK_MUTABLE_ORDERING_FIELD_INVALID": (
+ "test_pr3_rejects_invalid_ordering_fields",
+ "testPinnedMutableOrderingConfiguration",
+ ),
+ "FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED": (
+ "test_pr3_rejects_unsafe_index_configuration",
+ "testPinnedMutableIndexConfiguration",
+ ),
+ "FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED": (
+ "test_pr3_requires_index_bootstrap",
+ "testPinnedMutableIndexBootstrapConfiguration",
+ ),
}
SHA1_PATTERN = re.compile(r"[0-9a-f]{40}")
SHA256_PATTERN = re.compile(r"[0-9a-f]{64}")
@@ -272,13 +328,15 @@ def validate_manifest(manifest: dict[str, Any]) ->
list[str]:
)
if "value" not in setting:
errors.append(f"required_effective_settings[{index}].value is
required")
- if setting.get("applies_to") != "pr2-append-only-cow":
+ if setting.get("applies_to") not in {
+ "pr2-append-only-cow",
+ "pr3-mutable-cow",
+ }:
errors.append(
- f"required_effective_settings[{index}].applies_to must be "
- "'pr2-append-only-cow'"
+ f"required_effective_settings[{index}].applies_to is
unsupported"
)
if settings != EXPECTED_REQUIRED_SETTINGS:
- errors.append("required_effective_settings does not match the bounded
PR2 path")
+ errors.append("required_effective_settings does not match the bounded
paths")
fixture_artifacts = manifest.get("fixture_artifacts")
if not isinstance(fixture_artifacts, dict):
@@ -311,11 +369,40 @@ def validate_manifest(manifest: dict[str, Any]) ->
list[str]:
if executable_contract != expected_contract:
errors.append("executable_contract does not match the bounded PR2
path")
+ mutable_executable_contract = manifest.get("mutable_executable_contract")
+ expected_mutable_contract = {
+ "contract_schema": 2,
+ "table_lifecycle": "new",
+ "writer_model": "single_writer",
+ "external_catalog": False,
+ "mutability": "mutable",
+ "source_changelog_mode": "UPSERT",
+ "execution_mode": "STREAMING",
+ "require_checkpointing": True,
+ "record_key_mode": "stable_key",
+ "allow_replay_behaviors": [
+ "cannot_occur",
+ "duplicates_acceptable",
+ "must_collapse",
+ ],
+ "ordering_mode": "event_time",
+ "ordering_field_types": ["BIGINT", "TIMESTAMP"],
+ "reserved_ordering_field_values": ["no_precombine"],
+ "allow_emits_deletes": [False, True],
+ "delete_payload_when_deletes": "FULL_ROW",
+ "delete_payload_without_deletes": "NOT_APPLICABLE",
+ "partition_fields_mutable": False,
+ }
+ if mutable_executable_contract != expected_mutable_contract:
+ errors.append(
+ "mutable_executable_contract does not match the bounded PR3 path"
+ )
+
physical_types = manifest.get("physical_types")
if not isinstance(physical_types, dict):
errors.append("physical_types must be a table")
elif physical_types != EXPECTED_PHYSICAL_TYPES:
- errors.append("physical_types does not match the bounded PR2 scalar
surface")
+ errors.append("physical_types does not match the bounded scalar
surface")
else:
for type_group in ("simple", "parameterized"):
values = physical_types.get(type_group)
@@ -372,7 +459,7 @@ def validate_manifest(manifest: dict[str, Any]) ->
list[str]:
}
if implemented_evidence != EXPECTED_ACCEPTANCE_EVIDENCE:
errors.append(
- "implemented_acceptance_checks does not match the required PR2
evidence"
+ "implemented_acceptance_checks does not match the required
evidence"
)
return errors
@@ -384,13 +471,23 @@ def supported_option_keys(manifest: dict[str, Any]) ->
set[str]:
return {option["key"] for option in manifest["verified_options"]}
-def validation_evidence(manifest: dict[str, Any]) -> dict[str, Any]:
+def validation_evidence(
+ manifest: dict[str, Any], contract_schema: int | None = None
+) -> dict[str, Any]:
"""Build the stable baseline evidence included in every Flink
assessment."""
return {
"baseline_id": manifest["baseline_id"],
"capability_manifest_schema": manifest["schema_version"],
- "design_contract_schema":
manifest["executable_contract"]["contract_schema"],
+ "design_contract_schema": (
+ manifest["executable_contract"]["contract_schema"]
+ if contract_schema is None
+ else contract_schema
+ ),
+ "supported_design_contract_schemas": [
+ manifest["executable_contract"]["contract_schema"],
+ manifest["mutable_executable_contract"]["contract_schema"],
+ ],
"flink_fixture_version": manifest["flink"]["fixture_version"],
"hudi_source_revision": manifest["hudi"]["source_revision"],
}
diff --git a/hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py
b/hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py
index fe1e9a09c4e2..5a564fc08c2b 100644
--- a/hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py
+++ b/hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py
@@ -15,12 +15,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-"""Validate and render the bounded Hudi Architect Flink PR2 design contract.
+"""Validate and render bounded Hudi Architect Flink SQL design contracts.
The caller supplies every architecture decision. This program does not select
-table type, identity, partitioning, replay behavior, or checkpoint cadence. It
-only rejects incomplete or unsupported combinations and serializes a validated
-contract into deterministic Flink SQL.
+table type, identity, ordering, partitioning, replay behavior, or checkpoint
+cadence. It only rejects incomplete or unsupported combinations and serializes
+a validated contract into deterministic Flink SQL.
"""
from __future__ import annotations
@@ -41,7 +41,12 @@ from validate_flink_capabilities import (
validation_evidence,
)
-CONTRACT_SCHEMA_VERSION = 1
+APPEND_CONTRACT_SCHEMA_VERSION = 1
+MUTABLE_CONTRACT_SCHEMA_VERSION = 2
+SUPPORTED_CONTRACT_SCHEMAS = {
+ APPEND_CONTRACT_SCHEMA_VERSION,
+ MUTABLE_CONTRACT_SCHEMA_VERSION,
+}
MAX_TYPE_LENGTH = 2_147_483_647
STATUS_PRIORITY = {"INCOMPLETE": 0, "REVIEW_REQUIRED": 1, "BLOCKED": 2}
AZURE_FILESYSTEM_SCHEMES = frozenset({"abfs", "abfss"})
@@ -78,7 +83,15 @@ FINDING_STATUS = {
"FLINK_LOAD_BEARING_VALUE_REQUIRED": "INCOMPLETE",
"FLINK_MULTI_WRITER_REVIEW": "REVIEW_REQUIRED",
"FLINK_MUTABILITY_REQUIRED": "INCOMPLETE",
+ "FLINK_MUTABLE_AUTO_KEY_UNSUPPORTED": "BLOCKED",
"FLINK_MUTABLE_COW_DEFERRED": "BLOCKED",
+ "FLINK_MUTABLE_DELETE_PAYLOAD_UNSUPPORTED": "BLOCKED",
+ "FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED": "BLOCKED",
+ "FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED": "BLOCKED",
+ "FLINK_MUTABLE_ORDERING_FIELD_INVALID": "BLOCKED",
+ "FLINK_MUTABLE_ORDERING_FIELD_REQUIRED": "INCOMPLETE",
+ "FLINK_MUTABLE_PARTITION_EVOLUTION_UNSUPPORTED": "BLOCKED",
+ "FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED": "BLOCKED",
"FLINK_OPTION_NOT_VERIFIED": "REVIEW_REQUIRED",
"FLINK_PARTITION_FIELD_MISSING": "BLOCKED",
"FLINK_PHYSICAL_SCHEMA_REQUIRED": "INCOMPLETE",
@@ -97,6 +110,7 @@ FINDING_STATUS = {
"FLINK_VERSION_UNVERIFIED": "REVIEW_REQUIRED",
"FLINK_WRITER_MODEL_UNRESOLVED": "REVIEW_REQUIRED",
"FLINK_PR2_WRITE_PATH_UNSUPPORTED": "BLOCKED",
+ "FLINK_PR3_WRITE_PATH_UNSUPPORTED": "BLOCKED",
"FLINK_SCHEMA_TYPE_UNVERIFIED": "REVIEW_REQUIRED",
}
@@ -181,7 +195,7 @@ def _normalize_type(
]:
assessment.add(
"FLINK_SCHEMA_TYPE_UNVERIFIED",
- f"Flink SQL type {value!r} is outside the bounded PR2 scalar
surface",
+ f"Flink SQL type {value!r} is outside the bounded scalar surface",
)
return None
@@ -355,6 +369,7 @@ def _render_sql(
source = contract["source"]
runtime = contract["runtime"]
identity = contract["identity"]
+ mutable = contract["contract_schema"] == MUTABLE_CONTRACT_SCHEMA_VERSION
definitions = [
f" {_quote_identifier(column['name'])} {column['type']}"
@@ -373,15 +388,33 @@ def _render_sql(
if partition_fields:
partition_sql = ", ".join(_quote_identifier(field) for field in
partition_fields)
table_ddl += f"\nPARTITIONED BY ({partition_sql})"
- table_ddl += (
- "\nWITH (\n"
- " 'connector' = 'hudi',\n"
- f" 'path' = {_sql_string(table['path'])},\n"
- " 'table.type' = 'COPY_ON_WRITE',\n"
- " 'write.operation' = 'insert',\n"
- " 'write.insert.cluster' = 'false'\n"
- ");"
- )
+ if mutable:
+ ordering_field = contract["ordering"]["fields"][0]
+ table_ddl += (
+ "\nWITH (\n"
+ " 'connector' = 'hudi',\n"
+ f" 'path' = {_sql_string(table['path'])},\n"
+ " 'table.type' = 'COPY_ON_WRITE',\n"
+ " 'write.operation' = 'upsert',\n"
+ f" 'ordering.fields' = {_sql_string(ordering_field)},\n"
+ " 'hoodie.write.record.merge.mode' = 'EVENT_TIME_ORDERING',\n"
+ " 'index.type' = 'FLINK_STATE',\n"
+ " 'index.global.enabled' = 'true',\n"
+ " 'index.state.ttl' = '0',\n"
+ " 'index.bootstrap.enabled' = 'true',\n"
+ " 'changelog.enabled' = 'false'\n"
+ ");"
+ )
+ else:
+ table_ddl += (
+ "\nWITH (\n"
+ " 'connector' = 'hudi',\n"
+ f" 'path' = {_sql_string(table['path'])},\n"
+ " 'table.type' = 'COPY_ON_WRITE',\n"
+ " 'write.operation' = 'insert',\n"
+ " 'write.insert.cluster' = 'false'\n"
+ ");"
+ )
field_sql = ",\n ".join(_quote_identifier(column["name"]) for column in
table_columns)
insert_sql = (
@@ -405,7 +438,7 @@ def _render_sql(
def assess_design(
contract: dict[str, Any], manifest: dict[str, Any]
) -> dict[str, Any]:
- """Validate one explicit PR2 contract and return deterministic evidence."""
+ """Validate one explicit append-only or mutable COW contract."""
assessment = Assessment()
manifest_errors = validate_manifest(manifest)
@@ -420,7 +453,7 @@ def assess_design(
"findings": assessment.findings,
"advisory_codes": assessment.advisories,
"validation_evidence": {
- "design_contract_schema": CONTRACT_SCHEMA_VERSION,
+ "design_contract_schema": contract.get("contract_schema"),
},
}
@@ -432,6 +465,7 @@ def assess_design(
"flink_version",
"hudi_version",
"identity",
+ "ordering",
"runtime",
"safety",
"source",
@@ -441,10 +475,24 @@ def assess_design(
"contract",
assessment,
)
- if contract.get("contract_schema") != CONTRACT_SCHEMA_VERSION:
+ contract_schema = contract.get("contract_schema")
+ if (
+ not isinstance(contract_schema, int)
+ or isinstance(contract_schema, bool)
+ or contract_schema not in SUPPORTED_CONTRACT_SCHEMAS
+ ):
assessment.add(
"FLINK_DESIGN_CONTRACT_INVALID",
- f"contract_schema must be {CONTRACT_SCHEMA_VERSION}",
+ "contract_schema must be 1 (append-only) or 2 (mutable COW)",
+ )
+ mutable_contract = contract_schema == MUTABLE_CONTRACT_SCHEMA_VERSION
+ contract_profile = manifest[
+ "mutable_executable_contract" if mutable_contract else
"executable_contract"
+ ]
+ if not mutable_contract and "ordering" in contract:
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "ordering is supported only by contract_schema 2",
)
if contract.get("baseline_id") != manifest.get("baseline_id"):
assessment.add("FLINK_VERSION_UNVERIFIED", "The capability baseline
does not match")
@@ -475,14 +523,20 @@ def assess_design(
if lifecycle is None:
assessment.add("FLINK_TABLE_LIFECYCLE_REQUIRED", "Table lifecycle is
required")
elif lifecycle == "existing":
- assessment.add("FLINK_EXISTING_TABLE_DEFERRED", "PR2 supports only a
new table")
+ assessment.add(
+ "FLINK_EXISTING_TABLE_DEFERRED",
+ "The executable contracts support only a new Architect-managed
table",
+ )
elif lifecycle != "new":
assessment.add("FLINK_TABLE_LIFECYCLE_REQUIRED", "Table lifecycle is
unresolved")
writer_model = safety.get("writer_model")
if writer_model is None:
assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is
required")
elif writer_model == "multi_writer":
- assessment.add("FLINK_MULTI_WRITER_REVIEW", "PR2 supports only a
confirmed single writer")
+ assessment.add(
+ "FLINK_MULTI_WRITER_REVIEW",
+ "The executable contracts support only a confirmed single writer",
+ )
elif writer_model != "single_writer":
assessment.add("FLINK_WRITER_MODEL_UNRESOLVED", "Writer topology is
unresolved")
external_catalog = safety.get("external_catalog")
@@ -491,20 +545,32 @@ def assess_design(
elif external_catalog is not False:
assessment.add("FLINK_CATALOG_REQUIREMENT_UNRESOLVED", "Catalog
requirement is unresolved")
mutability = safety.get("mutability")
- if mutability == "mutable":
- assessment.add("FLINK_MUTABLE_COW_DEFERRED", "Mutable COW is deferred")
+ if mutable_contract:
+ if mutability != "mutable":
+ assessment.add(
+ "FLINK_MUTABILITY_REQUIRED",
+ "contract_schema 2 requires mutable input",
+ )
+ elif mutability == "mutable":
+ assessment.add(
+ "FLINK_MUTABLE_COW_DEFERRED",
+ "Mutable COW requires the explicit contract_schema 2 safety
contract",
+ )
elif mutability != "append_only":
assessment.add("FLINK_MUTABILITY_REQUIRED", "Append-only input must be
confirmed")
replay_behavior = safety.get("replay_behavior")
if not _non_empty_string(replay_behavior):
assessment.add("FLINK_REPLAY_BEHAVIOR_UNRESOLVED", "Replay behavior is
unresolved")
- elif replay_behavior == "must_collapse":
+ elif replay_behavior == "must_collapse" and not mutable_contract:
assessment.add("FLINK_REPLAY_IDEMPOTENCE_DEFERRED", "Replay
deduplication needs upsert")
- elif replay_behavior not in {"cannot_occur", "duplicates_acceptable"}:
+ elif replay_behavior not in contract_profile["allow_replay_behaviors"]:
assessment.add("FLINK_REPLAY_BEHAVIOR_UNRESOLVED", "Replay behavior is
unresolved")
table = _dict(contract.get("table"), "table", assessment)
- _check_keys(table, {"columns", "name", "partition_fields", "path"},
"table", assessment)
+ table_keys = {"columns", "name", "partition_fields", "path"}
+ if mutable_contract:
+ table_keys.add("partition_fields_mutable")
+ _check_keys(table, table_keys, "table", assessment)
table_name = table.get("name")
table_path = table.get("path")
if not _check_concrete_value(table_name) or any(
@@ -529,6 +595,17 @@ def assess_design(
partition_fields = _string_list(
table.get("partition_fields"), "table.partition_fields", assessment
)
+ if mutable_contract:
+ if table.get("partition_fields_mutable") is True:
+ assessment.add(
+ "FLINK_MUTABLE_PARTITION_EVOLUTION_UNSUPPORTED",
+ "The first mutable path does not move a key between
partitions",
+ )
+ elif table.get("partition_fields_mutable") is not False:
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "table.partition_fields_mutable must be explicitly false",
+ )
for field in partition_fields:
if field not in table_by_name:
assessment.add(
@@ -600,7 +677,8 @@ def assess_design(
f"Record-key option field {field!r} is absent from the
physical schema",
)
if identity_mode == "stable_key":
- assessment.advise("FLINK_STABLE_KEY_NOT_IDEMPOTENT")
+ if not mutable_contract:
+ assessment.advise("FLINK_STABLE_KEY_NOT_IDEMPOTENT")
if "auto_key_accepted" in identity:
assessment.add(
"FLINK_DESIGN_CONTRACT_INVALID",
@@ -639,7 +717,12 @@ def assess_design(
"The auto-key path cannot contain primary-key or record-key
fields",
)
accepted = identity.get("auto_key_accepted")
- if accepted is True:
+ if mutable_contract:
+ assessment.add(
+ "FLINK_MUTABLE_AUTO_KEY_UNSUPPORTED",
+ "Mutable upsert/delete requires a stable non-null record key",
+ )
+ elif accepted is True:
assessment.advise("FLINK_AUTO_KEY_DURABILITY")
elif accepted is False:
assessment.add("FLINK_AUTO_KEY_DECLINED", "Auto-generated keys
were declined")
@@ -651,8 +734,57 @@ def assess_design(
else:
assessment.add("FLINK_DESIGN_CONTRACT_INVALID", "identity.mode is
required")
+ ordering_fields: list[str] = []
+ if mutable_contract:
+ ordering = _dict(contract.get("ordering"), "ordering", assessment)
+ _check_keys(ordering, {"fields", "mode"}, "ordering", assessment)
+ ordering_fields = _string_list(
+ ordering.get("fields"), "ordering.fields", assessment
+ )
+ if ordering.get("mode") != "event_time":
+ assessment.add(
+ "FLINK_MUTABLE_ORDERING_FIELD_REQUIRED",
+ "The first mutable path requires event_time ordering",
+ )
+ if len(ordering_fields) != 1:
+ assessment.add(
+ "FLINK_MUTABLE_ORDERING_FIELD_REQUIRED",
+ "Exactly one ordering field is required",
+ )
+ reserved_ordering_values = set(
+ contract_profile["reserved_ordering_field_values"]
+ )
+ for field in ordering_fields:
+ if field in reserved_ordering_values:
+ assessment.add(
+ "FLINK_MUTABLE_ORDERING_FIELD_INVALID",
+ f"Ordering field {field!r} is reserved by Hudi and
disables ordering",
+ )
+ continue
+ column = table_by_name.get(field)
+ if column is None:
+ assessment.add(
+ "FLINK_MUTABLE_ORDERING_FIELD_INVALID",
+ f"Ordering field {field!r} is absent from the physical
schema",
+ )
+ elif column["nullable"]:
+ assessment.add(
+ "FLINK_MUTABLE_ORDERING_FIELD_INVALID",
+ f"Ordering field {field!r} must be NOT NULL",
+ )
+ elif column["type"] != "BIGINT" and re.fullmatch(
+ r"TIMESTAMP\([0-6]\)", column["type"]
+ ) is None:
+ assessment.add(
+ "FLINK_MUTABLE_ORDERING_FIELD_INVALID",
+ f"Ordering field {field!r} must use BIGINT or
TIMESTAMP(p<=6)",
+ )
+
source = _dict(contract.get("source"), "source", assessment)
- _check_keys(source, {"changelog_mode", "columns", "table"}, "source",
assessment)
+ source_keys = {"changelog_mode", "columns", "table"}
+ if mutable_contract:
+ source_keys.update({"delete_payload", "emits_deletes",
"primary_key_fields"})
+ _check_keys(source, source_keys, "source", assessment)
source_name = source.get("table")
if not _check_concrete_value(source_name) or any(
not part for part in str(source_name).split(".")
@@ -673,36 +805,74 @@ def assess_design(
"FLINK_SOURCE_SCHEMA_MISMATCH",
f"Source field {name!r} must match the target name, type,
and nullability",
)
+ if mutable_contract:
+ source_primary_key_fields = _string_list(
+ source.get("primary_key_fields"), "source.primary_key_fields",
assessment
+ )
+ if not source_primary_key_fields:
+ assessment.add(
+ "FLINK_SOURCE_CONTRACT_REQUIRED",
+ "The mutable source must declare its primary-key fields",
+ )
+ elif record_key_fields and source_primary_key_fields !=
record_key_fields:
+ assessment.add(
+ "FLINK_PRIMARY_KEY_RECORD_KEY_CONFLICT",
+ "Source primary-key fields must exactly match target
record-key fields",
+ )
+ for field in source_primary_key_fields:
+ source_column = source_by_name.get(field)
+ if source_column is None or source_column["nullable"]:
+ assessment.add(
+ "FLINK_PRIMARY_KEY_RECORD_KEY_CONFLICT",
+ f"Source primary-key field {field!r} must exist and be NOT
NULL",
+ )
+ emits_deletes = source.get("emits_deletes")
+ if not isinstance(emits_deletes, bool):
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "source.emits_deletes must be explicitly true or false",
+ )
+ delete_payload = source.get("delete_payload")
+ if emits_deletes is True and delete_payload != "FULL_ROW":
+ assessment.add(
+ "FLINK_MUTABLE_DELETE_PAYLOAD_UNSUPPORTED",
+ "Deletes must carry key, ordering, partition, and projected
row values",
+ )
+ elif emits_deletes is False and delete_payload != "NOT_APPLICABLE":
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "source.delete_payload must be NOT_APPLICABLE when deletes are
absent",
+ )
+
changelog_mode = source.get("changelog_mode")
if not _non_empty_string(changelog_mode) or changelog_mode == "UNKNOWN":
assessment.add(
"FLINK_SOURCE_CONTRACT_REQUIRED",
"The source changelog mode must be explicitly confirmed",
)
- elif changelog_mode != "INSERT_ONLY":
+ elif mutable_contract and changelog_mode != "UPSERT":
+ assessment.add(
+ "FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED",
+ "The mutable source must declare normalized UPSERT changelog
(I/UA/D, no UB)",
+ )
+ elif not mutable_contract and changelog_mode != "INSERT_ONLY":
assessment.add(
"FLINK_SOURCE_CHANGELOG_NOT_APPEND_ONLY",
"The PR2 source contract must be INSERT_ONLY",
)
write = _dict(contract.get("write"), "write", assessment)
+ write_keys = {"connector_options", "operation", "table_type"}
+ if mutable_contract:
+ write_keys.update({"changelog_enabled", "index", "record_merge_mode"})
+ else:
+ write_keys.add("insert_cluster")
_check_keys(
write,
- {"connector_options", "insert_cluster", "operation", "table_type"},
+ write_keys,
"write",
assessment,
)
- insert_cluster = write.get("insert_cluster")
- if insert_cluster is True:
- assessment.add(
- "FLINK_APPEND_MODE_CLUSTERING_ENABLED",
- "COW insert is append mode only with write.insert.cluster=false",
- )
- elif insert_cluster is not False:
- assessment.add(
- "FLINK_DESIGN_CONTRACT_INVALID",
- "write.insert_cluster must be explicitly true or false",
- )
table_type = write.get("table_type")
operation = write.get("operation")
if table_type is None or operation is None:
@@ -710,11 +880,70 @@ def assess_design(
"FLINK_DESIGN_CONTRACT_INVALID",
"write.table_type and write.operation must be explicit",
)
- elif table_type != "COPY_ON_WRITE" or operation != "insert":
+ elif mutable_contract and (
+ table_type != "COPY_ON_WRITE" or operation != "upsert"
+ ):
+ assessment.add(
+ "FLINK_PR3_WRITE_PATH_UNSUPPORTED",
+ "The first mutable path supports only COPY_ON_WRITE upsert",
+ )
+ elif not mutable_contract and (
+ table_type != "COPY_ON_WRITE" or operation != "insert"
+ ):
assessment.add(
"FLINK_PR2_WRITE_PATH_UNSUPPORTED",
"PR2 supports only COPY_ON_WRITE with write.operation=insert",
)
+ if mutable_contract:
+ if write.get("record_merge_mode") != "EVENT_TIME_ORDERING":
+ assessment.add(
+ "FLINK_PR3_WRITE_PATH_UNSUPPORTED",
+ "write.record_merge_mode must be EVENT_TIME_ORDERING",
+ )
+ if write.get("changelog_enabled") is not False:
+ assessment.add(
+ "FLINK_PR3_WRITE_PATH_UNSUPPORTED",
+ "write.changelog_enabled must be explicitly false",
+ )
+ index = _dict(write.get("index"), "write.index", assessment)
+ _check_keys(
+ index,
+ {"bootstrap_enabled", "global", "state_ttl_days", "type"},
+ "write.index",
+ assessment,
+ )
+ if index.get("type") != "FLINK_STATE" or index.get("global") is not
True:
+ assessment.add(
+ "FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED",
+ "The first mutable path requires global FLINK_STATE index",
+ )
+ state_ttl = index.get("state_ttl_days")
+ if (
+ not isinstance(state_ttl, (int, float))
+ or isinstance(state_ttl, bool)
+ or state_ttl != 0
+ ):
+ assessment.add(
+ "FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED",
+ "write.index.state_ttl_days must be 0 (no state expiry)",
+ )
+ if index.get("bootstrap_enabled") is not True:
+ assessment.add(
+ "FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED",
+ "FLINK_STATE must bootstrap existing keys after state loss or
cold restart",
+ )
+ else:
+ insert_cluster = write.get("insert_cluster")
+ if insert_cluster is True:
+ assessment.add(
+ "FLINK_APPEND_MODE_CLUSTERING_ENABLED",
+ "COW insert is append mode only with
write.insert.cluster=false",
+ )
+ elif insert_cluster is not False:
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "write.insert_cluster must be explicitly true or false",
+ )
connector_options = write.get("connector_options")
if not isinstance(connector_options, dict):
assessment.add("FLINK_DESIGN_CONTRACT_INVALID",
"write.connector_options must be an object")
@@ -726,7 +955,7 @@ def assess_design(
f"Options outside the pinned allowlist: {unknown_options}",
)
raw_insert_cluster = connector_options.get("write.insert.cluster")
- if "write.insert.cluster" in connector_options and not (
+ if not mutable_contract and "write.insert.cluster" in
connector_options and not (
raw_insert_cluster is False
or (
isinstance(raw_insert_cluster, str)
@@ -739,21 +968,27 @@ def assess_design(
)
raw_table_type = connector_options.get("table.type")
raw_operation = connector_options.get("write.operation")
+ expected_operation = "upsert" if mutable_contract else "insert"
if (
"table.type" in connector_options
and raw_table_type != "COPY_ON_WRITE"
) or (
- "write.operation" in connector_options and raw_operation !=
"insert"
+ "write.operation" in connector_options
+ and raw_operation != expected_operation
):
assessment.add(
- "FLINK_PR2_WRITE_PATH_UNSUPPORTED",
- "A connector-option override leaves the bounded COW insert
path",
+ (
+ "FLINK_PR3_WRITE_PATH_UNSUPPORTED"
+ if mutable_contract
+ else "FLINK_PR2_WRITE_PATH_UNSUPPORTED"
+ ),
+ "A connector-option override leaves the bounded COW path",
)
verified_overrides = sorted(set(connector_options) -
set(unknown_options))
if verified_overrides:
assessment.add(
"FLINK_DESIGN_CONTRACT_INVALID",
- "PR2 connector settings must use canonical structured fields; "
+ "Connector settings must use canonical structured fields; "
f"pass-through overrides are not rendered:
{verified_overrides}",
)
@@ -775,7 +1010,14 @@ def assess_design(
"FLINK_DESIGN_CONTRACT_INVALID", "runtime.execution_mode must be
explicit"
)
elif execution_mode != "STREAMING":
- assessment.add("FLINK_PR2_WRITE_PATH_UNSUPPORTED", "PR2 supports
streaming execution")
+ assessment.add(
+ (
+ "FLINK_PR3_WRITE_PATH_UNSUPPORTED"
+ if mutable_contract
+ else "FLINK_PR2_WRITE_PATH_UNSUPPORTED"
+ ),
+ "The executable path supports streaming execution",
+ )
checkpointing_enabled = runtime.get("checkpointing_enabled")
if checkpointing_enabled is False:
assessment.add("FLINK_CHECKPOINTING_REQUIRED", "Streaming commits
require checkpointing")
@@ -794,7 +1036,7 @@ def assess_design(
"FLINK_CHECKPOINT_INTERVAL_UNSUPPORTED",
"The checkpoint interval must be at least "
f"{manifest['runtime_constraints']['checkpoint_interval_min_ms']}
ms for the "
- "bounded PR2 path",
+ "bounded executable path",
)
target_freshness = runtime.get("target_commit_freshness_ms")
if target_freshness is not None and (
@@ -815,7 +1057,7 @@ def assess_design(
"finding_codes": [finding["code"] for finding in assessment.findings],
"findings": assessment.findings,
"advisory_codes": assessment.advisories,
- "validation_evidence": validation_evidence(manifest),
+ "validation_evidence": validation_evidence(manifest, contract_schema),
}
if executable:
result["artifacts"] = _render_sql(
diff --git
a/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr1_scenarios.toml
b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr1_scenarios.toml
index 8d3579a38648..d017169fc92a 100644
--- a/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr1_scenarios.toml
+++ b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr1_scenarios.toml
@@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-schema_version = 3
+schema_version = 4
status_precedence = ["INCOMPLETE", "REVIEW_REQUIRED", "BLOCKED"]
[auto_key_acceptance]
@@ -105,12 +105,13 @@ executable_eligible = false
[[scenarios]]
case_id = "F07_MUTABLE"
-finding_codes = ["FLINK_MUTABLE_COW_DEFERRED"]
-expected_final_status = "BLOCKED"
-executable_eligible = false
+all_other_gates_pass = true
+passes_safety_gates = true
+next_contract_schema = 2
[[scenarios]]
case_id = "F08_REPLAY_COLLAPSE"
+inputs = { mutability = "append_only" }
finding_codes = ["FLINK_REPLAY_IDEMPOTENCE_DEFERRED"]
expected_final_status = "BLOCKED"
executable_eligible = false
diff --git
a/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr3/mutable_cow.json
b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr3/mutable_cow.json
new file mode 100644
index 000000000000..0fcbbc84a812
--- /dev/null
+++
b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr3/mutable_cow.json
@@ -0,0 +1,66 @@
+{
+ "contract_schema": 2,
+ "baseline_id": "hudi-1.2.0-flink-1.20",
+ "hudi_version": "1.2.0",
+ "flink_version": "1.20.1",
+ "safety": {
+ "table_lifecycle": "new",
+ "writer_model": "single_writer",
+ "external_catalog": false,
+ "mutability": "mutable",
+ "replay_behavior": "must_collapse"
+ },
+ "table": {
+ "name": "orders_mutable_hudi",
+ "path": "file:///tmp/hudi-architect-mutable-orders",
+ "columns": [
+ {"name": "id", "type": "STRING", "nullable": false},
+ {"name": "name", "type": "STRING", "nullable": true},
+ {"name": "event_version", "type": "BIGINT", "nullable": false},
+ {"name": "partition_date", "type": "DATE", "nullable": false}
+ ],
+ "partition_fields": ["partition_date"],
+ "partition_fields_mutable": false
+ },
+ "identity": {
+ "mode": "stable_key",
+ "record_key_fields": ["id"],
+ "record_key_option_fields": []
+ },
+ "ordering": {
+ "mode": "event_time",
+ "fields": ["event_version"]
+ },
+ "source": {
+ "table": "orders_mutable_source",
+ "columns": [
+ {"name": "id", "type": "STRING", "nullable": false},
+ {"name": "name", "type": "STRING", "nullable": true},
+ {"name": "event_version", "type": "BIGINT", "nullable": false},
+ {"name": "partition_date", "type": "DATE", "nullable": false}
+ ],
+ "primary_key_fields": ["id"],
+ "changelog_mode": "UPSERT",
+ "emits_deletes": true,
+ "delete_payload": "FULL_ROW"
+ },
+ "write": {
+ "table_type": "COPY_ON_WRITE",
+ "operation": "upsert",
+ "record_merge_mode": "EVENT_TIME_ORDERING",
+ "index": {
+ "type": "FLINK_STATE",
+ "global": true,
+ "state_ttl_days": 0,
+ "bootstrap_enabled": true
+ },
+ "changelog_enabled": false,
+ "connector_options": {}
+ },
+ "runtime": {
+ "execution_mode": "STREAMING",
+ "checkpointing_enabled": true,
+ "checkpoint_interval_ms": 60000,
+ "target_commit_freshness_ms": 120000
+ }
+}
diff --git
a/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr3/mutable_cow.sql
b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr3/mutable_cow.sql
new file mode 100644
index 000000000000..4a90e0e42624
--- /dev/null
+++ b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr3/mutable_cow.sql
@@ -0,0 +1,55 @@
+/*
+ * 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.
+ */
+
+SET 'execution.runtime-mode' = 'streaming';
+SET 'execution.checkpointing.interval' = '60000 ms';
+
+CREATE TABLE `orders_mutable_hudi` (
+ `id` STRING NOT NULL,
+ `name` STRING,
+ `event_version` BIGINT NOT NULL,
+ `partition_date` DATE NOT NULL,
+ PRIMARY KEY (`id`) NOT ENFORCED
+)
+PARTITIONED BY (`partition_date`)
+WITH (
+ 'connector' = 'hudi',
+ 'path' = 'file:///tmp/hudi-architect-mutable-orders',
+ 'table.type' = 'COPY_ON_WRITE',
+ 'write.operation' = 'upsert',
+ 'ordering.fields' = 'event_version',
+ 'hoodie.write.record.merge.mode' = 'EVENT_TIME_ORDERING',
+ 'index.type' = 'FLINK_STATE',
+ 'index.global.enabled' = 'true',
+ 'index.state.ttl' = '0',
+ 'index.bootstrap.enabled' = 'true',
+ 'changelog.enabled' = 'false'
+);
+
+INSERT INTO `orders_mutable_hudi` (
+ `id`,
+ `name`,
+ `event_version`,
+ `partition_date`
+)
+SELECT
+ `id`,
+ `name`,
+ `event_version`,
+ `partition_date`
+FROM `orders_mutable_source`;
diff --git a/hudi-agent-gateway/tests/test_hudi_architect_skill.py
b/hudi-agent-gateway/tests/test_hudi_architect_skill.py
index a327bd4c00da..b662da9e4a4d 100644
--- a/hudi-agent-gateway/tests/test_hudi_architect_skill.py
+++ b/hudi-agent-gateway/tests/test_hudi_architect_skill.py
@@ -40,6 +40,7 @@ FLINK_VALIDATOR = SKILL_DIR / "validate_flink_capabilities.py"
FLINK_DESIGN_VALIDATOR = SKILL_DIR / "validate_flink_design.py"
PINNED_HUDI_REVISION = "f05c83f2b97732de7a558ff9b26959e1139c05f5"
PR2_FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" /
"flink_pr2"
+PR3_FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" /
"flink_pr3"
ASF_LICENSE_MARKER = "Licensed to the Apache Software Foundation (ASF) under
one"
PR2_JAVA_FIXTURE = (
GATEWAY_DIR
@@ -152,7 +153,7 @@ def
test_engine_router_is_lazy_and_keeps_spark_on_shared_flow() -> None:
assert "FLINK_" not in _read(REFERENCES_DIR / "config-templates.md")
-def test_flink_baseline_is_fixed_and_pr2_has_bounded_executable_contract() ->
None:
+def test_flink_baseline_has_bounded_append_and_mutable_contracts() -> None:
capability = _read(REFERENCES_DIR /
"flink-1.20-hudi-1.2.0-capabilities.md")
manifest = _load_toml(CAPABILITY_MANIFEST)
decisions = _read(REFERENCES_DIR / "flink-decision-overrides.md")
@@ -169,7 +170,7 @@ def
test_flink_baseline_is_fixed_and_pr2_has_bounded_executable_contract() -> No
"fixture_version": "1.20.1",
}
assert PINNED_HUDI_REVISION in capability
- assert manifest["schema_version"] == 2
+ assert manifest["schema_version"] == 3
assert manifest["fixture_artifacts"] == {
"hudi_bundle": "org.apache.hudi:hudi-flink1.20-bundle:1.2.0",
"flink_version": "1.20.1",
@@ -214,10 +215,11 @@ def
test_flink_capability_validation_emits_pinned_evidence() -> None:
assert result.returncode == 0, result.stderr
assert json.loads(result.stdout) == {
"baseline_id": "hudi-1.2.0-flink-1.20",
- "capability_manifest_schema": 2,
+ "capability_manifest_schema": 3,
"design_contract_schema": 1,
"flink_fixture_version": "1.20.1",
"hudi_source_revision": PINNED_HUDI_REVISION,
+ "supported_design_contract_schemas": [1, 2],
}
@@ -301,7 +303,7 @@ def
test_flink_capability_validation_rejects_executable_contract_drift(
assert expected_message in result.stderr
-def test_flink_capability_manifest_records_pr2_acceptance_contract() -> None:
+def test_flink_capability_manifest_records_executable_acceptance_contracts()
-> None:
manifest = _load_toml(CAPABILITY_MANIFEST)
python_tests = {
node.name
@@ -310,7 +312,11 @@ def
test_flink_capability_manifest_records_pr2_acceptance_contract() -> None:
}
java_fixture = _read(PR2_JAVA_FIXTURE)
effective_settings = {
- setting["key"]: setting["value"]
+ setting["applies_to"]: {
+ candidate["key"]: candidate["value"]
+ for candidate in manifest["required_effective_settings"]
+ if candidate["applies_to"] == setting["applies_to"]
+ }
for setting in manifest["required_effective_settings"]
}
implemented_checks = {
@@ -318,10 +324,30 @@ def
test_flink_capability_manifest_records_pr2_acceptance_contract() -> None:
for check in manifest["implemented_acceptance_checks"]
}
+ assert manifest["mutable_executable_contract"]["allow_replay_behaviors"]
== [
+ "cannot_occur",
+ "duplicates_acceptable",
+ "must_collapse",
+ ]
+ assert manifest["mutable_executable_contract"][
+ "reserved_ordering_field_values"
+ ] == ["no_precombine"]
assert effective_settings == {
- "table.type": "COPY_ON_WRITE",
- "write.insert.cluster": False,
- "write.operation": "insert",
+ "pr2-append-only-cow": {
+ "table.type": "COPY_ON_WRITE",
+ "write.insert.cluster": False,
+ "write.operation": "insert",
+ },
+ "pr3-mutable-cow": {
+ "changelog.enabled": False,
+ "hoodie.write.record.merge.mode": "EVENT_TIME_ORDERING",
+ "index.bootstrap.enabled": True,
+ "index.global.enabled": True,
+ "index.state.ttl": 0.0,
+ "index.type": "FLINK_STATE",
+ "table.type": "COPY_ON_WRITE",
+ "write.operation": "upsert",
+ },
}
assert implemented_checks == {
"FLINK_APPEND_MODE_CLUSTERING_ENABLED": (
@@ -356,6 +382,22 @@ def
test_flink_capability_manifest_records_pr2_acceptance_contract() -> None:
"test_pr2_enforces_checkpoint_interval_safety_floor",
"testPinnedRuntimeCheckpointIntervalBoundary",
),
+ "FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED": (
+ "test_pr3_rejects_unsupported_source_changelog",
+ "testMutableCowSinkAndUpsertPlan",
+ ),
+ "FLINK_MUTABLE_ORDERING_FIELD_INVALID": (
+ "test_pr3_rejects_invalid_ordering_fields",
+ "testPinnedMutableOrderingConfiguration",
+ ),
+ "FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED": (
+ "test_pr3_rejects_unsafe_index_configuration",
+ "testPinnedMutableIndexConfiguration",
+ ),
+ "FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED": (
+ "test_pr3_requires_index_bootstrap",
+ "testPinnedMutableIndexBootstrapConfiguration",
+ ),
}
for check in manifest["implemented_acceptance_checks"]:
assert check["python_test"] in python_tests
@@ -376,9 +418,9 @@ def
test_pinned_source_hashes_match_when_release_object_is_available() -> None:
assert result.returncode == 0, result.stderr
-def test_pr1_safety_gates_keep_deterministic_status_before_pr2_validation() ->
None:
+def
test_pr1_safety_gates_keep_deterministic_status_before_executable_validation()
-> None:
contract = _load_toml(GATE_FIXTURE)
- assert contract["schema_version"] == 3
+ assert contract["schema_version"] == 4
assert contract["status_precedence"] == ["INCOMPLETE", "REVIEW_REQUIRED",
"BLOCKED"]
precedence = {
status: priority for priority, status in
enumerate(contract["status_precedence"])
@@ -401,6 +443,14 @@ def
test_pr1_safety_gates_keep_deterministic_status_before_pr2_validation() -> N
assert resolved_status == scenario["expected_final_status"],
scenario["case_id"]
assert scenario["executable_eligible"] is False
+ mutable = next(
+ scenario
+ for scenario in contract["scenarios"]
+ if scenario["case_id"] == "F07_MUTABLE"
+ )
+ assert mutable["passes_safety_gates"] is True
+ assert mutable["next_contract_schema"] == 2
+
def test_combined_gate_fixture_preserves_all_reasons_and_blocked_wins() ->
None:
contract = _load_toml(GATE_FIXTURE)
@@ -459,12 +509,210 @@ def
test_pr2_valid_designs_render_exact_executable_sql(fixture_name: str) -> Non
).rstrip("\n")
assert assessment["validation_evidence"] == {
"baseline_id": "hudi-1.2.0-flink-1.20",
- "capability_manifest_schema": 2,
+ "capability_manifest_schema": 3,
"design_contract_schema": 1,
"flink_fixture_version": "1.20.1",
"hudi_source_revision": PINNED_HUDI_REVISION,
+ "supported_design_contract_schemas": [1, 2],
+ }
+
+
+def test_pr3_valid_mutable_design_renders_exact_executable_sql() -> None:
+ result = _run_flink_design_validator(PR3_FIXTURE_DIR / "mutable_cow.json")
+
+ assert result.returncode == 0, result.stderr
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "CONFIG_VALIDATED"
+ assert assessment["executable_eligible"] is True
+ assert assessment["finding_codes"] == []
+ assert assessment["advisory_codes"] == []
+ assert assessment["artifacts"]["combined_sql"] == _read_sql_golden(
+ PR3_FIXTURE_DIR / "mutable_cow.sql"
+ ).rstrip("\n")
+ assert assessment["validation_evidence"]["design_contract_schema"] == 2
+ assert assessment["validation_evidence"]["capability_manifest_schema"] == 3
+
+
[email protected](
+ "replay_behavior",
+ ["cannot_occur", "duplicates_acceptable", "must_collapse"],
+)
+def test_pr3_accepts_resolved_replay_behaviors(
+ tmp_path: Path, replay_behavior: str
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["safety"]["replay_behavior"] = replay_behavior
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 0, result.stderr
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "CONFIG_VALIDATED"
+ assert assessment["finding_codes"] == []
+ assert assessment["artifacts"]["combined_sql"] == _read_sql_golden(
+ PR3_FIXTURE_DIR / "mutable_cow.sql"
+ ).rstrip("\n")
+
+
[email protected]("replay_behavior", ["unknown", [], {}])
+def test_pr3_rejects_unresolved_replay_behaviors(
+ tmp_path: Path, replay_behavior: object
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["safety"]["replay_behavior"] = replay_behavior
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == ["FLINK_REPLAY_BEHAVIOR_UNRESOLVED"]
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert "artifacts" not in assessment
+
+
+def test_pr3_accepts_update_only_source_without_delete_payload(tmp_path: Path)
-> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["source"]["emits_deletes"] = False
+ design["source"]["delete_payload"] = "NOT_APPLICABLE"
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 0, result.stderr
+ assessment = json.loads(result.stdout)
+ assert assessment["status"] == "CONFIG_VALIDATED"
+ assert assessment["finding_codes"] == []
+
+
[email protected]("changelog_mode", ["INSERT_ONLY", "RETRACT", "ALL"])
+def test_pr3_rejects_unsupported_source_changelog(
+ tmp_path: Path, changelog_mode: str
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["source"]["changelog_mode"] = changelog_mode
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == [
+ "FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED"
+ ]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
+
[email protected](
+ ("field_name", "field_type", "nullable"),
+ [
+ ("event_version", "STRING", False),
+ ("event_version", "TIMESTAMP_LTZ(3)", False),
+ ("event_version", "BIGINT", True),
+ ("no_precombine", "BIGINT", False),
+ ],
+)
+def test_pr3_rejects_invalid_ordering_fields(
+ tmp_path: Path, field_name: str, field_type: str, nullable: bool
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ for section in ("table", "source"):
+ column = next(
+ column
+ for column in design[section]["columns"]
+ if column["name"] == "event_version"
+ )
+ column["name"] = field_name
+ column["type"] = field_type
+ column["nullable"] = nullable
+ design["ordering"]["fields"] = [field_name]
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert "FLINK_MUTABLE_ORDERING_FIELD_INVALID" in
assessment["finding_codes"]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
+
[email protected](
+ ("field", "value"),
+ [("type", "BUCKET"), ("global", False), ("state_ttl_days", 1)],
+)
+def test_pr3_rejects_unsafe_index_configuration(
+ tmp_path: Path, field: str, value: object
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["write"]["index"][field] = value
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == [
+ "FLINK_MUTABLE_INDEX_CONFIGURATION_UNSUPPORTED"
+ ]
+ assert assessment["status"] == "BLOCKED"
+
+
+def test_pr3_requires_index_bootstrap(tmp_path: Path) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["write"]["index"]["bootstrap_enabled"] = False
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] ==
["FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED"]
+ assert assessment["status"] == "BLOCKED"
+
+
+def test_pr3_rejects_table_property_alias_for_write_merge_mode(tmp_path: Path)
-> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ design["write"]["connector_options"] = {
+ "hoodie.record.merge.mode": "EVENT_TIME_ORDERING"
}
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == ["FLINK_OPTION_NOT_VERIFIED"]
+ assert assessment["status"] == "REVIEW_REQUIRED"
+ assert "artifacts" not in assessment
+
+
[email protected](
+ ("mutation", "expected_code"),
+ [
+ (("identity", "mode", "auto_key"),
"FLINK_MUTABLE_AUTO_KEY_UNSUPPORTED"),
+ (
+ ("table", "partition_fields_mutable", True),
+ "FLINK_MUTABLE_PARTITION_EVOLUTION_UNSUPPORTED",
+ ),
+ (
+ ("source", "delete_payload", "KEY_ONLY"),
+ "FLINK_MUTABLE_DELETE_PAYLOAD_UNSUPPORTED",
+ ),
+ (("write", "operation", "insert"), "FLINK_PR3_WRITE_PATH_UNSUPPORTED"),
+ ],
+)
+def test_pr3_fails_closed_outside_the_bounded_mutable_path(
+ tmp_path: Path, mutation: tuple[str, str, object], expected_code: str
+) -> None:
+ design = json.loads(_read(PR3_FIXTURE_DIR / "mutable_cow.json"))
+ section, field, value = mutation
+ design[section][field] = value
+ if section == "identity" and field == "mode":
+ design["identity"]["record_key_fields"] = []
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert expected_code in assessment["finding_codes"]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
def test_pr2_fails_closed_when_capability_manifest_is_malformed(tmp_path:
Path) -> None:
manifest_path = tmp_path / "capabilities.toml"
@@ -486,6 +734,22 @@ def
test_pr2_fails_closed_when_capability_manifest_is_malformed(tmp_path: Path)
assert "artifacts" not in assessment
[email protected]("contract_schema", [True, 1.0, 3])
+def test_flink_design_rejects_non_exact_contract_schema(
+ tmp_path: Path, contract_schema: object
+) -> None:
+ design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
+ design["contract_schema"] = contract_schema
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert "FLINK_DESIGN_CONTRACT_INVALID" in assessment["finding_codes"]
+ assert assessment["executable_eligible"] is False
+ assert "artifacts" not in assessment
+
+
def test_pr2_rejects_record_key_missing_from_append_schema(tmp_path: Path) ->
None:
design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
design["write"]["connector_options"] = {
@@ -1052,7 +1316,7 @@ def test_safety_gate_order_is_stable() -> None:
"## F7 — Physical table and source contract",
"## F8 — Source changelog contract",
"## F9 — Streaming checkpoint contract",
- "## PR2 validation and completion",
+ "## PR2/PR3 validation and completion",
]
offsets = [flow.index(heading) for heading in headings]
assert offsets == sorted(offsets)
