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 a99e89ed2835 feat(agent): Add first executable mutable Flink SQL MOR
sink path to Hudi Architect (#20302)
a99e89ed2835 is described below
commit a99e89ed28351cb1e2f78bed7b48fbbd9e0c37c6
Author: fhan <[email protected]>
AuthorDate: Sat Oct 17 09:50:28 2026 +0800
feat(agent): Add first executable mutable Flink SQL MOR sink path to Hudi
Architect (#20302)
* feat(agent): Add the first executable mutable Flink SQL MOR sink path to
Hudi Architect
* fix(agent): harden Flink MOR contract validation
---------
Co-authored-by: fhan <[email protected]>
---
.github/workflows/agent_gateway_ci.yml | 3 +
.../fixtures/hudi-architect-flink-1.20/pom.xml | 6 +
.../architect/TestFlinkArchitectSqlFixtures.java | 82 ++++++
hudi-agent-gateway/skills/hudi-architect/README.md | 16 +-
.../skills/hudi-architect/RUNBOOK.md | 11 +-
hudi-agent-gateway/skills/hudi-architect/SKILL.md | 20 +-
.../flink-1.20-hudi-1.2.0-capabilities.md | 38 ++-
.../flink-1.20-hudi-1.2.0-capabilities.toml | 195 +++++++++++++-
.../references/flink-config-templates.md | 51 +++-
.../references/flink-decision-overrides.md | 41 ++-
.../references/flink-question-flow.md | 76 +++++-
.../hudi-architect/references/flink-warnings.md | 83 +++++-
.../hudi-architect/validate_flink_capabilities.py | 118 ++++++++-
.../skills/hudi-architect/validate_flink_design.py | 290 ++++++++++++++++++---
.../hudi_architect/flink_pr4/mutable_mor.json | 85 ++++++
.../hudi_architect/flink_pr4/mutable_mor.sql | 62 +++++
.../tests/test_hudi_architect_skill.py | 261 ++++++++++++++++++-
17 files changed, 1340 insertions(+), 98 deletions(-)
diff --git a/.github/workflows/agent_gateway_ci.yml
b/.github/workflows/agent_gateway_ci.yml
index aafccf8b6936..47e31b8f34df 100644
--- a/.github/workflows/agent_gateway_ci.yml
+++ b/.github/workflows/agent_gateway_ci.yml
@@ -74,6 +74,9 @@ jobs:
python skills/hudi-architect/validate_flink_design.py \
--input tests/fixtures/hudi_architect/flink_pr3/mutable_cow.json \
--emit-sql > /dev/null
+ python skills/hudi-architect/validate_flink_design.py \
+ --input tests/fixtures/hudi_architect/flink_pr4/mutable_mor.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 0714ab8d6fbf..919f08edf04c 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
@@ -88,6 +88,12 @@
<include>*.sql</include>
</includes>
</testResource>
+ <testResource>
+
<directory>${project.basedir}/../../tests/fixtures/hudi_architect/flink_pr4</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 89ef68792fb7..fe2a074bf238 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
@@ -143,12 +143,39 @@ class TestFlinkArchitectSqlFixtures {
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);
+ assertFalse(plan.contains("compact_plan_generate"), plan);
+ assertFalse(plan.contains("compact_task"), plan);
+ assertFalse(plan.contains("compact_commit"), plan);
assertEquals(
Set.of(RowKind.INSERT, RowKind.UPDATE_AFTER, RowKind.DELETE),
ChangelogModes.UPSERT.getContainedKinds());
assertFalse(ChangelogModes.UPSERT.contains(RowKind.UPDATE_BEFORE));
}
+ @Test
+ void testMutableMorSinkAndCompactionPlan() throws Exception {
+ TestContext context = newTestContext();
+ StreamTableEnvironment tableEnv = context.tableEnv;
+ registerMutableSource(context);
+
+ SqlFixture fixture = SqlFixture.load("mutable_mor.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_mor_hudi"), plan);
+ assertTrue(plan.contains("stream_write:
default_database.orders_mutable_mor_hudi"), plan);
+ assertTrue(plan.contains("changelogMode=[I,UA,D]"), plan);
+ assertTrue(plan.contains("index_bootstrap"), plan);
+ assertTrue(plan.contains("compact_plan_generate"), plan);
+ assertTrue(plan.contains("compact_task"), plan);
+ assertTrue(plan.contains("compact_commit"), plan);
+ }
+
@Test
void testPinnedMutableOrderingConfiguration() {
Configuration conf = mutableConfiguration();
@@ -179,6 +206,48 @@ class TestFlinkArchitectSqlFixtures {
assertTrue(conf.get(FlinkOptions.INDEX_BOOTSTRAP_ENABLED));
}
+ @Test
+ void testPinnedMorCompactionConfiguration() {
+ Configuration conf = mutableMorConfiguration();
+
+ assertTrue(OptionsResolver.isMorTable(conf));
+ assertTrue(OptionsResolver.needsScheduleCompaction(conf));
+ assertTrue(OptionsResolver.needsAsyncCompaction(conf));
+ assertTrue(conf.get(FlinkOptions.COMPACTION_SCHEDULE_ENABLED));
+ assertTrue(conf.get(FlinkOptions.COMPACTION_ASYNC_ENABLED));
+
assertTrue(conf.get(FlinkOptions.COMPACTION_OPERATION_EXECUTE_ASYNC_ENABLED));
+ assertEquals(FlinkOptions.NUM_COMMITS,
conf.get(FlinkOptions.COMPACTION_TRIGGER_STRATEGY));
+ assertEquals(5, conf.get(FlinkOptions.COMPACTION_DELTA_COMMITS));
+ }
+
+ @Test
+ void testPinnedMorClusteringDisabled() {
+ Configuration conf = mutableMorConfiguration();
+
+ assertFalse(conf.get(FlinkOptions.CLUSTERING_SCHEDULE_ENABLED));
+ assertFalse(conf.get(FlinkOptions.CLUSTERING_ASYNC_ENABLED));
+ }
+
+ @Test
+ void testPinnedMorQuerySemantics() {
+ Configuration conf = mutableMorConfiguration();
+
+ assertEquals("snapshot", FlinkOptions.QUERY_TYPE_SNAPSHOT);
+ assertEquals("read_optimized", FlinkOptions.QUERY_TYPE_READ_OPTIMIZED);
+ assertEquals(FlinkOptions.QUERY_TYPE_SNAPSHOT,
conf.get(FlinkOptions.QUERY_TYPE));
+ conf.set(FlinkOptions.QUERY_TYPE, FlinkOptions.QUERY_TYPE_READ_OPTIMIZED);
+ assertEquals(FlinkOptions.QUERY_TYPE_READ_OPTIMIZED,
conf.get(FlinkOptions.QUERY_TYPE));
+ }
+
+ @Test
+ void testPinnedAppendOnlyMorDoesNotScheduleCompaction() {
+ Configuration conf = mutableMorConfiguration();
+ conf.set(FlinkOptions.OPERATION, "insert");
+
+ assertTrue(OptionsResolver.isAppendMode(conf));
+ assertFalse(OptionsResolver.needsScheduleCompaction(conf));
+ }
+
@Test
void testPinnedFactorySkipsMissingRecordKeyCheckInAppendMode() {
TestContext context = newTestContext();
@@ -432,6 +501,19 @@ class TestFlinkArchitectSqlFixtures {
return conf;
}
+ private static Configuration mutableMorConfiguration() {
+ Configuration conf = mutableConfiguration();
+ conf.set(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
+ conf.set(FlinkOptions.COMPACTION_SCHEDULE_ENABLED, true);
+ conf.set(FlinkOptions.COMPACTION_ASYNC_ENABLED, true);
+ conf.set(FlinkOptions.COMPACTION_OPERATION_EXECUTE_ASYNC_ENABLED, true);
+ conf.set(FlinkOptions.COMPACTION_TRIGGER_STRATEGY,
FlinkOptions.NUM_COMMITS);
+ conf.set(FlinkOptions.COMPACTION_DELTA_COMMITS, 5);
+ conf.set(FlinkOptions.CLUSTERING_SCHEDULE_ENABLED, false);
+ conf.set(FlinkOptions.CLUSTERING_ASYNC_ENABLED, false);
+ return conf;
+ }
+
private static TestContext newTestContext() {
StreamExecutionEnvironment environment =
StreamExecutionEnvironment.getExecutionEnvironment();
environment.enableCheckpointing(60_000L);
diff --git a/hudi-agent-gateway/skills/hudi-architect/README.md
b/hudi-agent-gateway/skills/hudi-architect/README.md
index e4796e0818bc..11a603043b69 100644
--- a/hudi-agent-gateway/skills/hudi-architect/README.md
+++ b/hudi-agent-gateway/skills/hudi-architect/README.md
@@ -35,11 +35,12 @@ 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 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
+- **Flink** combines the routing and safety foundation with bounded
append-only COW, mutable COW,
+ and mutable MOR 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.
+ partitions, and the fixed global FLINK_STATE/bootstrap/no-TTL contract. MOR
additionally fixes
+ same-writer asynchronous compaction, records snapshot/read-optimized
freshness, and disables
+ clustering. Other Flink designs remain fail-closed.
## How to invoke
@@ -123,6 +124,8 @@ 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`.
+The mutable MOR schema-3 fixture is
+`hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr4/mutable_mor.json`.
Credential-bearing evidence can be sanitized without executing or parsing it
as configuration:
@@ -144,8 +147,9 @@ This is **Milestone 1 of a longer arc** — meant to be
shareable and playable,
## What's out of scope in Milestone 1
-- Flink MOR, existing-table takeover, external catalogs, multiple writers,
alternative indexes,
- custom mergers, retract normalization, partition movement, key-only deletes,
full CDC history,
+- Flink append-only MOR, standalone compaction, detailed compaction tuning,
clustering,
+ 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 SQL output as if it were generally
applicable.
diff --git a/hudi-agent-gateway/skills/hudi-architect/RUNBOOK.md
b/hudi-agent-gateway/skills/hudi-architect/RUNBOOK.md
index 5ab8c8d4d05b..688c4e810794 100644
--- a/hudi-agent-gateway/skills/hudi-architect/RUNBOOK.md
+++ b/hudi-agent-gateway/skills/hudi-architect/RUNBOOK.md
@@ -25,7 +25,8 @@ 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
-bounded append-only and first mutable new-table, single-writer,
no-external-catalog COW paths.
+bounded append-only COW, mutable COW, and mutable MOR new-table, single-writer,
+no-external-catalog paths.
---
@@ -144,7 +145,8 @@ surface reaches `CONFIG_VALIDATED` only after the physical
target and source sch
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`
+FLINK_STATE/bootstrap/no-TTL contract. Mutable MOR additionally requires
same-writer asynchronous
+compaction, explicit snapshot/read-optimized freshness expectations, and
disabled clustering. 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).
@@ -153,8 +155,9 @@ 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** — bounded append-only and first mutable
new-table, single-writer,
- no-external-catalog COW streaming SQL sinks are implemented. MOR,
existing-table takeover,
+- **Flink executable coverage** — bounded append-only COW, mutable COW, and
mutable MOR
+ new-table, single-writer, no-external-catalog streaming SQL sinks are
implemented. Append-only
+ MOR, standalone compaction, detailed compaction tuning, clustering,
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.
diff --git a/hudi-agent-gateway/skills/hudi-architect/SKILL.md
b/hudi-agent-gateway/skills/hudi-architect/SKILL.md
index bae7f9a110df..55654efc38e3 100644
--- a/hudi-agent-gateway/skills/hudi-architect/SKILL.md
+++ b/hudi-agent-gateway/skills/hudi-architect/SKILL.md
@@ -75,14 +75,16 @@ 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, 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.
+The Flink implementation includes the PR1 safety foundation and three bounded
executable paths:
+PR2 append-only COW, PR3 mutable COW, and PR4 mutable MOR. All require a new
table, confirmed
+single writer, no external catalog, streaming checkpointing, and
`CONFIG_VALIDATED` from
+`validate_flink_design.py`. Both mutable paths additionally require stable
non-null identity, one
+event-time ordering field, normalized UPSERT changelog, full-row deletes when
present, immutable
+partitions, and fixed global FLINK_STATE/bootstrap/no-TTL settings. The MOR
path further requires
+snapshot/read-optimized expectations, compaction owned by the same Flink
writer, fixed asynchronous
+five-delta-commit compaction, and clustering disabled. Append-only MOR,
standalone table services,
+detailed compaction tuning, existing-table takeover, catalog, multi-writer,
alternative indexes,
+custom mergers, retract normalization, and full CDC history still fail closed.
## Flow structure
@@ -91,7 +93,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 bounded validator-rendered PR2/PR3 SQL artifacts.
+ assessment, or bounded validator-rendered PR2/PR3/PR4 SQL artifacts.
Load `references/question-flow.md` for the full round-by-round question list
with conditional gating.
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 7aead7775916..e4d0ce50f81e 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
@@ -83,6 +83,7 @@ the current checkout contains it.
| Auto-generated-key durability warning | Supported |
| Replay-idempotence classification | Supported |
| First bounded mutable COW upsert/delete path | Supported after schema-2
validation |
+| First bounded mutable MOR upsert/delete path | Supported after schema-3
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`) |
@@ -91,7 +92,12 @@ the current checkout contains it.
| 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 |
+| COW/MOR derivation | Workload-derived for mutable paths; append-only remains
COW |
+| MOR compaction ownership | Same Flink writer only; standalone or unknown
ownership withholds SQL |
+| MOR compaction | Async schedule and execution, `num_commits`, five delta
commits |
+| MOR snapshot freshness | Base files plus log state at a visible completed
instant |
+| MOR read-optimized freshness | Compacted base-file state; current only after
compaction |
+| Detailed compaction tuning or clustering | `BLOCKED`; deferred to
operations/separate work |
| New-table, single-writer, append-only COW SQL | Supported after static
validation |
| Stable-key DDL | `PRIMARY KEY (...) NOT ENFORCED` |
| Explicitly accepted auto-generated keys | Supported with durable warning |
@@ -188,13 +194,41 @@ Bootstrap closes the cold-start gap for a table created
and operated under this
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.
+## Implemented PR4 mutable MOR contract
+
+Schema 3 reuses every schema-2 mutable safety invariant and changes only the
table type plus an
+explicit read/table-service contract. The skill derives MOR from sustained
mutable writes and
+reader needs; it does not ask the user to select a Hudi table type without
context. Low or
+occasional updates, simplest-operation priority, and strict read-optimized
freshness derive COW.
+Append-only MOR is rejected because Hudi 1.2.0 treats MOR insert as append
mode and does not enter
+the verified mutable compaction schedule path.
+
+The canonical MOR sink pins `MERGE_ON_READ` upsert, event-time ordering,
global `FLINK_STATE` with
+TTL 0 and bootstrap, and `changelog.enabled=false`. The same Flink writer owns
compaction;
+`compaction.schedule.enabled`, `compaction.async.enabled`, and
+`compaction.operation.execute.async.enabled` are true, with `num_commits` and
five delta commits.
+Both clustering controls remain false. Five is the Hudi 1.2.0 effective
default selected as a fixed
+bounded contract, not a workload-tuned SLA.
+
+Snapshot queries read base files plus log state at a visible completed
instant. Read-optimized
+queries read compacted base files and can lag delta commits until compaction.
The reader requirement
+is recorded in the design contract, but reader-side query options are not
emitted into sink DDL.
+A standalone compactor is treated as an independent writer and remains
review-required; unknown
+ownership also fails closed. Compaction parallelism, time triggers, target IO,
clustering,
+concurrency configuration, and live service execution are not claimed.
+
+The pinned Flink planner fixture consumes the exact schema-3 golden SQL and
verifies the mutable
+writer, index bootstrap, `compact_plan_generate`, `compact_task`, and
`compact_commit` operators.
+Focused release-bundle fixtures retain effective MOR, compaction, clustering,
and query-mode
+semantics without claiming that a live job ran.
+
## 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 applicable gate and all load-bearing PR2 or PR3
design values passed the
+- `CONFIG_VALIDATED` — every applicable gate and all load-bearing PR2, PR3, or
PR4 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 d193a36d97b6..b70a775e0c2b 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 = 3
+schema_version = 4
baseline_id = "hudi-1.2.0-flink-1.20"
validation_surface = "initial-flink-sql-sink"
@@ -62,6 +62,34 @@ delete_payload_when_deletes = "FULL_ROW"
delete_payload_without_deletes = "NOT_APPLICABLE"
partition_fields_mutable = false
+[mutable_mor_executable_contract]
+contract_schema = 3
+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
+table_type = "MERGE_ON_READ"
+operation = "upsert"
+compaction_owner = "flink_writer"
+compaction_mode = "async"
+compaction_trigger_strategy = "num_commits"
+compaction_delta_commits = 5
+query_modes = ["snapshot", "read_optimized"]
+read_optimized_freshness = "after_compaction"
+clustering_enabled = 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]
@@ -142,7 +170,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 and no-precombine ordering sentinel
resolution"
+purpose = "Effective append mode, MOR table type, compaction gates, and
no-precombine resolution"
[[source_anchors]]
id = "table_factory"
@@ -172,7 +200,19 @@ purpose = "Release behavior exercised by the existing
factory fixtures"
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"
+purpose = "Sink changelog-mode negotiation and mutable MOR compaction-pipeline
selection"
+
+[[source_anchors]]
+id = "table_source"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java"
+sha256 = "94819175df0f9b3f76a641f45f70f6c1eee0e2da3976c3f4ff4883a831618504"
+purpose = "Snapshot and read-optimized Flink SQL source construction"
+
+[[source_anchors]]
+id = "hoodie_source"
+path =
"hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java"
+sha256 = "1100c640aa61e8bf4f3bcd584b00d720388cbb6a801d864028c5bf98591ec7dc"
+purpose = "Snapshot and read-optimized MOR file-slice selection"
[[source_anchors]]
id = "changelog_modes"
@@ -190,7 +230,7 @@ purpose = "FLINK_STATE backend and zero-TTL state behavior"
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"
+purpose = "Index bootstrap and compaction operators in the Flink write
pipeline"
[[source_anchors]]
id = "bootstrap_operator"
@@ -218,6 +258,48 @@ value_type = "boolean"
source_anchors = ["flink_options", "table_sink", "changelog_modes"]
source_symbol = "CHANGELOG_ENABLED"
+[[verified_options]]
+key = "clustering.async.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options"]
+source_symbol = "CLUSTERING_ASYNC_ENABLED"
+
+[[verified_options]]
+key = "clustering.schedule.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options"]
+source_symbol = "CLUSTERING_SCHEDULE_ENABLED"
+
+[[verified_options]]
+key = "compaction.async.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options", "options_resolver", "table_sink",
"pipelines"]
+source_symbol = "COMPACTION_ASYNC_ENABLED"
+
+[[verified_options]]
+key = "compaction.delta_commits"
+value_type = "integer"
+source_anchors = ["flink_options"]
+source_symbol = "COMPACTION_DELTA_COMMITS"
+
+[[verified_options]]
+key = "compaction.operation.execute.async.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options", "table_sink", "pipelines"]
+source_symbol = "COMPACTION_OPERATION_EXECUTE_ASYNC_ENABLED"
+
+[[verified_options]]
+key = "compaction.schedule.enabled"
+value_type = "boolean"
+source_anchors = ["flink_options", "options_resolver"]
+source_symbol = "COMPACTION_SCHEDULE_ENABLED"
+
+[[verified_options]]
+key = "compaction.trigger.strategy"
+value_type = "string"
+source_anchors = ["flink_options"]
+source_symbol = "COMPACTION_TRIGGER_STRATEGY"
+
[[verified_options]]
key = "hoodie.datasource.write.partitionpath.field"
value_type = "string"
@@ -345,6 +427,81 @@ key = "write.operation"
value = "upsert"
applies_to = "pr3-mutable-cow"
+[[required_effective_settings]]
+key = "changelog.enabled"
+value = false
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "clustering.async.enabled"
+value = false
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "clustering.schedule.enabled"
+value = false
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "compaction.async.enabled"
+value = true
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "compaction.delta_commits"
+value = 5
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "compaction.operation.execute.async.enabled"
+value = true
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "compaction.schedule.enabled"
+value = true
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "compaction.trigger.strategy"
+value = "num_commits"
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "hoodie.write.record.merge.mode"
+value = "EVENT_TIME_ORDERING"
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "index.bootstrap.enabled"
+value = true
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "index.global.enabled"
+value = true
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "index.state.ttl"
+value = 0.0
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "index.type"
+value = "FLINK_STATE"
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "table.type"
+value = "MERGE_ON_READ"
+applies_to = "pr4-mutable-mor"
+
+[[required_effective_settings]]
+key = "write.operation"
+value = "upsert"
+applies_to = "pr4-mutable-mor"
+
[[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."
@@ -416,3 +573,33 @@ 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"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MOR_CLUSTERING_UNSUPPORTED"
+requirement = "Keep clustering disabled on the first mutable MOR path."
+python_test = "test_pr4_rejects_clustering"
+java_test = "testPinnedMorClusteringDisabled"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MOR_APPEND_ONLY_UNSUPPORTED"
+requirement = "Keep append-only writes on COW because MOR insert enters append
mode and does not schedule compaction."
+python_test = "test_pr4_rejects_append_only_mor"
+java_test = "testPinnedAppendOnlyMorDoesNotScheduleCompaction"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MOR_COMPACTION_CONFIGURATION_UNSUPPORTED"
+requirement = "Pin scheduling and async compaction to the same Flink writer
with a five-delta-commit trigger."
+python_test = "test_pr4_rejects_incompatible_compaction_configuration"
+java_test = "testPinnedMorCompactionConfiguration"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED"
+requirement = "Record that read-optimized queries expose compacted base-file
state rather than every delta commit."
+python_test = "test_pr4_rejects_strict_read_optimized_freshness"
+java_test = "testPinnedMorQuerySemantics"
+
+[[implemented_acceptance_checks]]
+id = "FLINK_PR4_WRITE_PATH_UNSUPPORTED"
+requirement = "Plan the exact mutable MOR SQL through the released connector
and retain its writer plus compaction operators."
+python_test = "test_pr4_valid_mutable_mor_design_renders_exact_executable_sql"
+java_test = "testMutableMorSinkAndCompactionPlan"
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 2759417f0934..f60212d613fb 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
@@ -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: 3
-Design contract schema: <1 append-only | 2 mutable COW>
+Capability manifest schema: 4
+Design contract schema: <1 append-only COW | 2 mutable COW | 3 mutable MOR>
Confirmed facts:
- <facts established by explicit answers>
@@ -65,8 +65,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: 3
-Design contract schema: <1 append-only | 2 mutable COW>
+Capability manifest schema: 4
+Design contract schema: <1 append-only COW | 2 mutable COW | 3 mutable MOR>
Confirmed facts:
- New table; confirmed single writer; no current external-catalog requirement.
@@ -77,16 +77,18 @@ Confirmed facts:
Source contract:
- Existing table: <source identifier>
- Expected physical schema: <Avro-compatible field names, bounded types,
nullability>
-- Changelog: <INSERT_ONLY for schema 1 | normalized UPSERT (I/UA/D, no UB) for
schema 2>
+- Changelog: <INSERT_ONLY for schema 1 | normalized UPSERT (I/UA/D, no UB) for
schemas 2/3>
- 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 for schema 1 | upsert for schema 2>
+- Table type: <COPY_ON_WRITE for schemas 1/2 | MERGE_ON_READ for schema 3>
+- Operation: <insert for schema 1 | upsert for schemas 2/3>
- Path-specific fixed settings: <insert clustering false | EVENT_TIME_ORDERING
plus global
FLINK_STATE, TTL 0, bootstrap true, and changelog false>
+- MOR only: compaction owned by this Flink writer, snapshot/read-optimized
freshness recorded,
+ fixed async five-delta-commit compaction, and clustering disabled.
Runtime contract:
- Execution: streaming
@@ -159,6 +161,37 @@ The mutable source must declare the same primary key as
the target and normalize
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.
+For schema 3 mutable MOR, preserve every schema-2 identity, ordering, source,
index, and runtime
+invariant, replace only `table.type`, and append the bounded service settings:
+
+```sql
+WITH (
+ 'connector' = 'hudi',
+ 'path' = '<concrete credential-free target path>',
+ 'table.type' = 'MERGE_ON_READ',
+ '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',
+ 'compaction.schedule.enabled' = 'true',
+ 'compaction.async.enabled' = 'true',
+ 'compaction.operation.execute.async.enabled' = 'true',
+ 'compaction.trigger.strategy' = 'num_commits',
+ 'compaction.delta_commits' = '5',
+ 'clustering.schedule.enabled' = 'false',
+ 'clustering.async.enabled' = 'false'
+)
+```
+
+Compaction is scheduled and executed by the same Flink writer job. Snapshot
readers include log
+state at a visible completed instant; read-optimized readers expose compacted
base files and are
+current only after compaction. Five delta commits is a trigger, not a time
SLA. Reader-side
+`hoodie.datasource.query.type` is intentionally absent from sink DDL.
+
## Deployment checks
Always retain these outside the static success claim:
@@ -176,8 +209,8 @@ Always retain these outside the static success claim:
- A complete Kafka, CDC, JDBC, or other source connector job.
- A Flink deployment or submit command.
- A Spark fallback.
-- OCC, NBCC, lock-provider, alternative index, catalog, MOR, compaction, or
clustering
- configuration.
+- OCC, NBCC, lock-provider, alternative index, catalog, append-only MOR,
standalone compaction,
+ detailed compaction tuning, or enabled clustering.
- 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
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 92dd6baae6cb..2faf994a2572 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
@@ -46,8 +46,8 @@ do not replace the shared table-design rules and do not form
a standalone planne
- 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 a
- complete version-1 append-only or version-2 mutable contract. Never
hand-write around a
- validator finding.
+ complete version-1 append-only COW, version-2 mutable COW, or version-3
mutable MOR contract.
+ Never hand-write around a validator finding.
## Gate outcomes
@@ -93,13 +93,21 @@ do not replace the shared table-design rules and do not
form a standalone planne
| 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 |
+| Append-only input requests MOR | `FLINK_MOR_APPEND_ONLY_UNSUPPORTED` |
`BLOCKED` | No |
+| MOR compaction owner is missing | `FLINK_MOR_COMPACTION_OWNER_REQUIRED` |
`INCOMPLETE` | No |
+| MOR compaction schedule, execution, mode, trigger, or threshold differs from
the bounded contract | `FLINK_MOR_COMPACTION_CONFIGURATION_UNSUPPORTED` |
`BLOCKED` | No |
+| MOR request supplies detailed compaction tuning |
`FLINK_MOR_COMPACTION_TUNING_UNSUPPORTED` | `BLOCKED` | No |
+| MOR reader mode or freshness requirement is unresolved |
`FLINK_MOR_QUERY_REQUIREMENT_UNRESOLVED` | `INCOMPLETE` | No |
+| MOR read-optimized results must include every completed write |
`FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED` | `BLOCKED` | No |
+| MOR enables clustering | `FLINK_MOR_CLUSTERING_UNSUPPORTED` | `BLOCKED` | No
|
+| Table type, operation, merge mode, changelog setting, compaction,
clustering, or execution mode is outside PR4 |
`FLINK_PR4_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 or PR3 contract passes deterministic validation | — |
`CONFIG_VALIDATED` | Yes |
+| Complete PR2, PR3, or PR4 contract passes deterministic validation | — |
`CONFIG_VALIDATED` | Yes |
## Final-status precedence
@@ -124,9 +132,9 @@ only bounded static validation against the pinned release
contract.
| Mutability | Stable business key | Replay requirement | Result |
|---|---|---|---|
-| Mutable | Yes | Replays impossible | Continue into the bounded PR3 mutable
contract |
+| Mutable | Yes | Replays impossible | Continue to COW/MOR derivation, then
schema 2 or 3 |
| Mutable | Yes | Duplicates acceptable | Continue; record duplicate tolerance
|
-| Mutable | Yes | Copies must collapse | Continue into the bounded PR3 mutable
contract |
+| Mutable | Yes | Copies must collapse | Continue to COW/MOR derivation, then
schema 2 or 3 |
| Mutable | Yes | Replay behavior unknown | `REVIEW_REQUIRED` |
| Mutable | No | Any | `BLOCKED` — auto-key cannot express mutable identity |
| Mutable | Unknown | Any | `INCOMPLETE` |
@@ -171,7 +179,7 @@ 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` | Otherwise-safe mutable workload enters the schema-2
candidate flow | Proceed to PR3 validation | Not decided by gates alone |
+| `F07_MUTABLE` | Otherwise-safe mutable workload enters table-type derivation
| Proceed to schema-2 or schema-3 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 |
@@ -222,6 +230,27 @@ to the executable contract just like `F10_SAFE_APPEND`.
- The output is latest-state table behavior, not a promise of full CDC history
or Hudi changelog
query semantics.
+## PR4 mutable MOR executable invariants
+
+- Version-3 contracts preserve every mutable schema-2 identity, ordering,
changelog, delete,
+ partition, index, checkpoint, and new-table safety invariant; only the table
type and explicit
+ MOR service/read contract differ.
+- The workload decision selects MOR only for sustained mutable writes when
snapshot readers are
+ acceptable or read-optimized readers explicitly accept compaction lag.
Append-only remains COW;
+ strict read-optimized freshness derives COW rather than an unsafe MOR
promise.
+- Canonical output fixes `MERGE_ON_READ` upsert, `EVENT_TIME_ORDERING`, global
`FLINK_STATE`, TTL
+ `0`, bootstrap enabled, and `changelog.enabled=false`.
+- The same Flink writer owns asynchronous compaction. Scheduling, async
execution, and async
+ compaction-operation execution are enabled with `num_commits` and five delta
commits. Five is a
+ fixed Hudi 1.2.0 effective default, not a tuned SLA.
+- A standalone compactor is an independent writer and remains
`REVIEW_REQUIRED`; concurrency
+ configuration is deferred to PR7. Unknown ownership also withholds SQL.
+- Snapshot readers see base plus log state at a visible completed instant.
Read-optimized readers
+ see compacted base-file state and advance only after compaction. Reader
`query.type` is not a
+ sink option and is not rendered into the DDL.
+- Clustering is disabled. Detailed compaction tuning, append-only MOR,
existing tables, catalogs,
+ multi-writer operation, and live service validation remain outside this path.
+
## 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 67468b7cac4b..93f49a84e92a 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,7 +140,7 @@ First ask whether existing logical records can change:
> - **Yes — updates or deletes occur**
> - **Not sure**
-- Mutable → enter the bounded PR3 candidate flow. Require a stable, non-null
business key and one
+- Mutable → enter the bounded mutable 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.
@@ -207,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 → continue only for an otherwise eligible mutable
schema-2 path. For the
- append-only schema-1 path, use `FLINK_REPLAY_IDEMPOTENCE_DEFERRED` with
`BLOCKED`.
+- Copies must collapse → continue only for an otherwise eligible mutable
schema-2 or schema-3
+ 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
@@ -298,16 +298,70 @@ State that a completed checkpoint coordinates the Hudi
commit. Do not claim end-
exactly-once: the source, checkpoint storage, restart behavior, and the rest
of the pipeline remain
deployment responsibilities. If known rate and active-partition evidence
indicates very little
data per checkpoint, surface `FLINK_CHECKPOINT_SMALL_FILE_RISK`; do not enable
clustering or tune
-file sizing in PR2.
+file sizing in the bounded executable paths.
-## PR2/PR3 validation and completion
+## F10 — Mutable table type, query freshness, and MOR ownership
-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.
+Ask this section only for an otherwise eligible mutable workload. Append-only
stays on schema 1
+COW and must not receive compaction questions.
+
+First establish write/read pressure in workload terms:
+
+> "Are updates sustained enough that rewriting whole data files is a concern,
and how will
+> consumers read the latest state?"
+>
+> - **Updates are low or occasional; prefer the simplest operations**
+> - **Sustained upserts; consumers use snapshot reads**
+> - **Sustained upserts; consumers use read-optimized scans and accept
compaction lag**
+> - **Read-optimized scans must include every completed write immediately**
+> - **Not sure**
+
+- Low/occasional updates or simplest operations → derive mutable COW and
schema 2.
+- Sustained upserts with snapshot reads → recommend MOR and continue the
schema-3 questions.
+- Sustained upserts with read-optimized consumers that accept compaction lag →
recommend MOR and
+ continue. Record `read_optimized_freshness=after_compaction`.
+- Read-optimized results required after every commit → recommend COW. If MOR
remains mandatory,
+ add `FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED` with `BLOCKED`.
+- Unknown query/freshness requirement →
`FLINK_MOR_QUERY_REQUIREMENT_UNRESOLVED` with `INCOMPLETE`;
+ do not emit MOR SQL.
+
+For a MOR candidate, record whether readers require snapshot, read-optimized,
or both. Snapshot
+reads expose base files plus log state at a visible completed instant.
Read-optimized reads expose
+compacted base-file state and therefore advance only after compaction. Do not
render reader-side
+`hoodie.datasource.query.type` into sink DDL, and do not turn the fixed
five-delta-commit trigger
+into a wall-clock freshness SLA.
+
+Then establish compaction ownership:
+
+> "Who will schedule and execute compaction for this table?"
+>
+> - **The same Flink writer job**
+> - **A separately deployed compaction service**
+> - **Not sure**
+
+- Same Flink writer → schema 3 may continue with asynchronous schedule and
execution enabled,
+ `num_commits`, five delta commits, and clustering disabled.
+- Standalone service → `REVIEW_REQUIRED` with `FLINK_MULTI_WRITER_REVIEW`;
emit no SQL. PR7 owns
+ the concurrency contract.
+- Not sure → `REVIEW_REQUIRED` with `FLINK_WRITER_MODEL_UNRESOLVED`; do not
guess ownership.
+- Missing owner in a schema-3 contract → `INCOMPLETE` with
+ `FLINK_MOR_COMPACTION_OWNER_REQUIRED`.
+
+Do not ask for compaction tasks, seconds, target IO, strategy alternatives, or
other tuning. If
+the request supplies such tuning, add
`FLINK_MOR_COMPACTION_TUNING_UNSUPPORTED`; operational
+tuning is outside this bounded design path. Clustering must remain disabled or
+`FLINK_MOR_CLUSTERING_UNSUPPORTED` blocks output.
+
+## PR2/PR3/PR4 validation and completion
+
+After all applicable questions, build a version-1 JSON contract for
append-only COW, a version-2
+contract for mutable COW, or a version-3 contract for mutable MOR. All contain
confirmed safety
+facts, physical schemas, identity, source, write, and runtime contracts.
Versions 2 and 3 record
+event-time ordering, normalized UPSERT, full-row delete posture, immutable
partition fields, and the fixed global
+FLINK_STATE/bootstrap/no-TTL contract. Version 3 additionally records query
modes and
+read-optimized freshness, compaction ownership and fixed schedule/execution
settings, and disabled
+clustering. Do not insert placeholders. Record an explicitly empty
+pass-through connector-option map; no 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 ef66e6f7bc89..5e277fd6f6ef 100644
--- a/hudi-agent-gateway/skills/hudi-architect/references/flink-warnings.md
+++ b/hudi-agent-gateway/skills/hudi-architect/references/flink-warnings.md
@@ -123,10 +123,10 @@ ordering, and partition references can eventually be
checked.
**Trigger:** Existing logical records can be updated or deleted, but the
request is still using the
version-1 append contract.
-**Message:** Mutable COW requires the explicit version-2 identity, ordering,
changelog, delete,
-partition, and index safety contract.
+**Message:** Mutable input requires the explicit version-2 COW or version-3
MOR identity, ordering,
+changelog, delete, partition, and index safety contract after table-type
derivation.
-**Action:** `BLOCKED`; qualify the request for schema 2 or generate no
executable configuration.
+**Action:** `BLOCKED`; qualify the request for schema 2 or 3 or generate no
executable configuration.
## FLINK_MUTABILITY_REQUIRED
@@ -195,7 +195,7 @@ 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`; use the mutable schema-2 contract only if all of its
gates are satisfied.
+**Action:** `BLOCKED`; use a mutable schema-2 or schema-3 contract only if all
gates are satisfied.
## FLINK_REPLAY_BEHAVIOR_UNRESOLVED
@@ -317,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. An eligible mutable request must use the separate schema-2
contract.
+append-only. An eligible mutable request must use a schema-2 COW or schema-3
MOR contract.
**Action:** `BLOCKED`; do not generate executable append SQL.
@@ -415,6 +415,79 @@ mutable configuration surface.
**Action:** `BLOCKED`; retain the request as a later capability rather than
guessing settings.
+## FLINK_MOR_APPEND_ONLY_UNSUPPORTED
+
+**Trigger:** An append-only contract requests a MOR table.
+
+**Message:** The bounded append route remains COW. Hudi 1.2.0 treats MOR
insert as append mode and
+does not schedule the mutable MOR compaction path that PR4 verifies.
+
+**Action:** `BLOCKED`; use schema-1 COW or reassess whether the workload is
actually mutable.
+
+## FLINK_MOR_COMPACTION_OWNER_REQUIRED
+
+**Trigger:** A schema-3 MOR contract omits compaction ownership.
+
+**Message:** The validator cannot assume that compaction runs inside the only
confirmed writer.
+
+**Action:** `INCOMPLETE`; identify the owner before emitting SQL.
+
+## FLINK_MOR_COMPACTION_CONFIGURATION_UNSUPPORTED
+
+**Trigger:** MOR schedule, async execution, operation execution, mode, trigger
strategy, or delta
+commit threshold differs from the bounded schema-3 contract.
+
+**Message:** PR4 verifies same-writer asynchronous compaction using
`num_commits` and five delta
+commits; it is not a general compaction planner.
+
+**Action:** `BLOCKED`; retain the fixed contract or defer to operational
design.
+
+## FLINK_MOR_COMPACTION_TUNING_UNSUPPORTED
+
+**Trigger:** A request supplies compaction task count, seconds, target IO,
timeout, or other
+detailed tuning.
+
+**Message:** These values depend on observed workload behavior and are outside
design-time static
+validation.
+
+**Action:** `BLOCKED`; defer tuning to operations rather than passing it
through silently.
+
+## FLINK_MOR_QUERY_REQUIREMENT_UNRESOLVED
+
+**Trigger:** Snapshot/read-optimized usage or the acceptable read-optimized
freshness is missing
+or unresolved.
+
+**Message:** MOR suitability depends on whether consumers read logs or can
accept compaction lag.
+
+**Action:** `INCOMPLETE`; obtain the reader contract before selecting schema 3.
+
+## FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED
+
+**Trigger:** Read-optimized consumers require every completed delta commit
immediately.
+
+**Message:** Read-optimized queries expose compacted base files; the
five-delta-commit trigger is
+not an immediate or time-based freshness guarantee.
+
+**Action:** `BLOCKED`; derive COW or explicitly accept freshness only after
compaction.
+
+## FLINK_MOR_CLUSTERING_UNSUPPORTED
+
+**Trigger:** Clustering schedule or async execution is enabled for schema 3.
+
+**Message:** PR4 owns only the minimum MOR compaction service and does not
compose clustering.
+
+**Action:** `BLOCKED`; keep both clustering controls disabled.
+
+## FLINK_PR4_WRITE_PATH_UNSUPPORTED
+
+**Trigger:** Mutable MOR table type, operation, merge mode, changelog,
compaction, clustering, or
+runtime leaves the bounded schema-3 path.
+
+**Message:** PR4 implements one conservative mutable MOR route, not the
complete Hudi table-service
+surface.
+
+**Action:** `BLOCKED`; do not guess or pass through unsupported 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 db9cf70cbc32..8038b0abc303 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 = 3
+EXPECTED_SCHEMA_VERSION = 4
EXPECTED_BASELINE_ID = "hudi-1.2.0-flink-1.20"
EXPECTED_HUDI_VERSION = "1.2.0"
EXPECTED_HUDI_SOURCE_REVISION = "f05c83f2b97732de7a558ff9b26959e1139c05f5"
@@ -102,6 +102,81 @@ EXPECTED_REQUIRED_SETTINGS = [
"value": "upsert",
"applies_to": "pr3-mutable-cow",
},
+ {
+ "key": "changelog.enabled",
+ "value": False,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "clustering.async.enabled",
+ "value": False,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "clustering.schedule.enabled",
+ "value": False,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "compaction.async.enabled",
+ "value": True,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "compaction.delta_commits",
+ "value": 5,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "compaction.operation.execute.async.enabled",
+ "value": True,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "compaction.schedule.enabled",
+ "value": True,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "compaction.trigger.strategy",
+ "value": "num_commits",
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "hoodie.write.record.merge.mode",
+ "value": "EVENT_TIME_ORDERING",
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "index.bootstrap.enabled",
+ "value": True,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "index.global.enabled",
+ "value": True,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "index.state.ttl",
+ "value": 0.0,
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "index.type",
+ "value": "FLINK_STATE",
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "table.type",
+ "value": "MERGE_ON_READ",
+ "applies_to": "pr4-mutable-mor",
+ },
+ {
+ "key": "write.operation",
+ "value": "upsert",
+ "applies_to": "pr4-mutable-mor",
+ },
]
EXPECTED_PHYSICAL_TYPES = {
"simple": [
@@ -196,6 +271,26 @@ EXPECTED_ACCEPTANCE_EVIDENCE = {
"test_pr3_requires_index_bootstrap",
"testPinnedMutableIndexBootstrapConfiguration",
),
+ "FLINK_MOR_CLUSTERING_UNSUPPORTED": (
+ "test_pr4_rejects_clustering",
+ "testPinnedMorClusteringDisabled",
+ ),
+ "FLINK_MOR_APPEND_ONLY_UNSUPPORTED": (
+ "test_pr4_rejects_append_only_mor",
+ "testPinnedAppendOnlyMorDoesNotScheduleCompaction",
+ ),
+ "FLINK_MOR_COMPACTION_CONFIGURATION_UNSUPPORTED": (
+ "test_pr4_rejects_incompatible_compaction_configuration",
+ "testPinnedMorCompactionConfiguration",
+ ),
+ "FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED": (
+ "test_pr4_rejects_strict_read_optimized_freshness",
+ "testPinnedMorQuerySemantics",
+ ),
+ "FLINK_PR4_WRITE_PATH_UNSUPPORTED": (
+ "test_pr4_valid_mutable_mor_design_renders_exact_executable_sql",
+ "testMutableMorSinkAndCompactionPlan",
+ ),
}
SHA1_PATTERN = re.compile(r"[0-9a-f]{40}")
SHA256_PATTERN = re.compile(r"[0-9a-f]{64}")
@@ -331,6 +426,7 @@ def validate_manifest(manifest: dict[str, Any]) ->
list[str]:
if setting.get("applies_to") not in {
"pr2-append-only-cow",
"pr3-mutable-cow",
+ "pr4-mutable-mor",
}:
errors.append(
f"required_effective_settings[{index}].applies_to is
unsupported"
@@ -398,6 +494,25 @@ def validate_manifest(manifest: dict[str, Any]) ->
list[str]:
"mutable_executable_contract does not match the bounded PR3 path"
)
+ mutable_mor_executable_contract =
manifest.get("mutable_mor_executable_contract")
+ expected_mutable_mor_contract = {
+ **expected_mutable_contract,
+ "contract_schema": 3,
+ "table_type": "MERGE_ON_READ",
+ "operation": "upsert",
+ "compaction_owner": "flink_writer",
+ "compaction_mode": "async",
+ "compaction_trigger_strategy": "num_commits",
+ "compaction_delta_commits": 5,
+ "query_modes": ["snapshot", "read_optimized"],
+ "read_optimized_freshness": "after_compaction",
+ "clustering_enabled": False,
+ }
+ if mutable_mor_executable_contract != expected_mutable_mor_contract:
+ errors.append(
+ "mutable_mor_executable_contract does not match the bounded PR4
path"
+ )
+
physical_types = manifest.get("physical_types")
if not isinstance(physical_types, dict):
errors.append("physical_types must be a table")
@@ -487,6 +602,7 @@ def validation_evidence(
"supported_design_contract_schemas": [
manifest["executable_contract"]["contract_schema"],
manifest["mutable_executable_contract"]["contract_schema"],
+ manifest["mutable_mor_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 5a564fc08c2b..37504d1215ef 100644
--- a/hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py
+++ b/hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py
@@ -43,9 +43,11 @@ from validate_flink_capabilities import (
APPEND_CONTRACT_SCHEMA_VERSION = 1
MUTABLE_CONTRACT_SCHEMA_VERSION = 2
+MUTABLE_MOR_CONTRACT_SCHEMA_VERSION = 3
SUPPORTED_CONTRACT_SCHEMAS = {
APPEND_CONTRACT_SCHEMA_VERSION,
MUTABLE_CONTRACT_SCHEMA_VERSION,
+ MUTABLE_MOR_CONTRACT_SCHEMA_VERSION,
}
MAX_TYPE_LENGTH = 2_147_483_647
STATUS_PRIORITY = {"INCOMPLETE": 0, "REVIEW_REQUIRED": 1, "BLOCKED": 2}
@@ -92,6 +94,13 @@ FINDING_STATUS = {
"FLINK_MUTABLE_ORDERING_FIELD_REQUIRED": "INCOMPLETE",
"FLINK_MUTABLE_PARTITION_EVOLUTION_UNSUPPORTED": "BLOCKED",
"FLINK_MUTABLE_SOURCE_CHANGELOG_UNSUPPORTED": "BLOCKED",
+ "FLINK_MOR_APPEND_ONLY_UNSUPPORTED": "BLOCKED",
+ "FLINK_MOR_CLUSTERING_UNSUPPORTED": "BLOCKED",
+ "FLINK_MOR_COMPACTION_CONFIGURATION_UNSUPPORTED": "BLOCKED",
+ "FLINK_MOR_COMPACTION_OWNER_REQUIRED": "INCOMPLETE",
+ "FLINK_MOR_COMPACTION_TUNING_UNSUPPORTED": "BLOCKED",
+ "FLINK_MOR_QUERY_REQUIREMENT_UNRESOLVED": "INCOMPLETE",
+ "FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED": "BLOCKED",
"FLINK_OPTION_NOT_VERIFIED": "REVIEW_REQUIRED",
"FLINK_PARTITION_FIELD_MISSING": "BLOCKED",
"FLINK_PHYSICAL_SCHEMA_REQUIRED": "INCOMPLETE",
@@ -111,6 +120,7 @@ FINDING_STATUS = {
"FLINK_WRITER_MODEL_UNRESOLVED": "REVIEW_REQUIRED",
"FLINK_PR2_WRITE_PATH_UNSUPPORTED": "BLOCKED",
"FLINK_PR3_WRITE_PATH_UNSUPPORTED": "BLOCKED",
+ "FLINK_PR4_WRITE_PATH_UNSUPPORTED": "BLOCKED",
"FLINK_SCHEMA_TYPE_UNVERIFIED": "REVIEW_REQUIRED",
}
@@ -369,7 +379,11 @@ def _render_sql(
source = contract["source"]
runtime = contract["runtime"]
identity = contract["identity"]
- mutable = contract["contract_schema"] == MUTABLE_CONTRACT_SCHEMA_VERSION
+ mutable = contract["contract_schema"] in {
+ MUTABLE_CONTRACT_SCHEMA_VERSION,
+ MUTABLE_MOR_CONTRACT_SCHEMA_VERSION,
+ }
+ mutable_mor = contract["contract_schema"] ==
MUTABLE_MOR_CONTRACT_SCHEMA_VERSION
definitions = [
f" {_quote_identifier(column['name'])} {column['type']}"
@@ -390,11 +404,12 @@ def _render_sql(
table_ddl += f"\nPARTITIONED BY ({partition_sql})"
if mutable:
ordering_field = contract["ordering"]["fields"][0]
+ table_type = "MERGE_ON_READ" if mutable_mor else "COPY_ON_WRITE"
table_ddl += (
"\nWITH (\n"
" 'connector' = 'hudi',\n"
f" 'path' = {_sql_string(table['path'])},\n"
- " 'table.type' = 'COPY_ON_WRITE',\n"
+ f" 'table.type' = '{table_type}',\n"
" 'write.operation' = 'upsert',\n"
f" 'ordering.fields' = {_sql_string(ordering_field)},\n"
" 'hoodie.write.record.merge.mode' = 'EVENT_TIME_ORDERING',\n"
@@ -402,9 +417,19 @@ def _render_sql(
" 'index.global.enabled' = 'true',\n"
" 'index.state.ttl' = '0',\n"
" 'index.bootstrap.enabled' = 'true',\n"
- " 'changelog.enabled' = 'false'\n"
- ");"
+ " 'changelog.enabled' = 'false'"
)
+ if mutable_mor:
+ table_ddl += (
+ ",\n 'compaction.schedule.enabled' = 'true',\n"
+ " 'compaction.async.enabled' = 'true',\n"
+ " 'compaction.operation.execute.async.enabled' = 'true',\n"
+ " 'compaction.trigger.strategy' = 'num_commits',\n"
+ " 'compaction.delta_commits' = '5',\n"
+ " 'clustering.schedule.enabled' = 'false',\n"
+ " 'clustering.async.enabled' = 'false'"
+ )
+ table_ddl += "\n);"
else:
table_ddl += (
"\nWITH (\n"
@@ -438,7 +463,7 @@ def _render_sql(
def assess_design(
contract: dict[str, Any], manifest: dict[str, Any]
) -> dict[str, Any]:
- """Validate one explicit append-only or mutable COW contract."""
+ """Validate one explicit append-only COW, mutable COW, or mutable MOR
contract."""
assessment = Assessment()
manifest_errors = validate_manifest(manifest)
@@ -466,34 +491,58 @@ def assess_design(
"hudi_version",
"identity",
"ordering",
+ "read",
"runtime",
"safety",
"source",
"table",
+ "table_services",
"write",
},
"contract",
assessment,
)
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
- ):
+ supported_contract_schema = (
+ isinstance(contract_schema, int)
+ and not isinstance(contract_schema, bool)
+ and contract_schema in SUPPORTED_CONTRACT_SCHEMAS
+ )
+ if not supported_contract_schema:
assessment.add(
"FLINK_DESIGN_CONTRACT_INVALID",
- "contract_schema must be 1 (append-only) or 2 (mutable COW)",
+ "contract_schema must be 1 (append-only), 2 (mutable COW), or 3
(mutable MOR)",
)
- mutable_contract = contract_schema == MUTABLE_CONTRACT_SCHEMA_VERSION
- contract_profile = manifest[
- "mutable_executable_contract" if mutable_contract else
"executable_contract"
- ]
+ mutable_contract = supported_contract_schema and contract_schema in {
+ MUTABLE_CONTRACT_SCHEMA_VERSION,
+ MUTABLE_MOR_CONTRACT_SCHEMA_VERSION,
+ }
+ mutable_mor_contract = (
+ supported_contract_schema
+ and contract_schema == MUTABLE_MOR_CONTRACT_SCHEMA_VERSION
+ )
+ if mutable_mor_contract:
+ contract_profile = manifest["mutable_mor_executable_contract"]
+ elif mutable_contract:
+ contract_profile = manifest["mutable_executable_contract"]
+ else:
+ contract_profile = manifest["executable_contract"]
if not mutable_contract and "ordering" in contract:
assessment.add(
"FLINK_DESIGN_CONTRACT_INVALID",
- "ordering is supported only by contract_schema 2",
+ "ordering is supported only by mutable contract schemas 2 and 3",
)
+ if not mutable_mor_contract:
+ if "read" in contract:
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "read is supported only by contract_schema 3",
+ )
+ if "table_services" in contract:
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "table_services is supported only by contract_schema 3",
+ )
if contract.get("baseline_id") != manifest.get("baseline_id"):
assessment.add("FLINK_VERSION_UNVERIFIED", "The capability baseline
does not match")
if not _non_empty_string(contract.get("hudi_version")) or not
_non_empty_string(
@@ -549,12 +598,12 @@ def assess_design(
if mutability != "mutable":
assessment.add(
"FLINK_MUTABILITY_REQUIRED",
- "contract_schema 2 requires mutable input",
+ "Mutable contract schemas 2 and 3 require mutable input",
)
elif mutability == "mutable":
assessment.add(
"FLINK_MUTABLE_COW_DEFERRED",
- "Mutable COW requires the explicit contract_schema 2 safety
contract",
+ "Mutable input requires an explicit mutable safety contract",
)
elif mutability != "append_only":
assessment.add("FLINK_MUTABILITY_REQUIRED", "Append-only input must be
confirmed")
@@ -865,6 +914,8 @@ def assess_design(
write_keys = {"connector_options", "operation", "table_type"}
if mutable_contract:
write_keys.update({"changelog_enabled", "index", "record_merge_mode"})
+ if mutable_mor_contract:
+ write_keys.add("clustering")
else:
write_keys.add("insert_cluster")
_check_keys(
@@ -880,7 +931,14 @@ def assess_design(
"FLINK_DESIGN_CONTRACT_INVALID",
"write.table_type and write.operation must be explicit",
)
- elif mutable_contract and (
+ elif mutable_mor_contract and (
+ table_type != "MERGE_ON_READ" or operation != "upsert"
+ ):
+ assessment.add(
+ "FLINK_PR4_WRITE_PATH_UNSUPPORTED",
+ "The first mutable MOR path supports only MERGE_ON_READ upsert",
+ )
+ elif mutable_contract and not mutable_mor_contract and (
table_type != "COPY_ON_WRITE" or operation != "upsert"
):
assessment.add(
@@ -890,19 +948,33 @@ def assess_design(
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 table_type == "MERGE_ON_READ":
+ assessment.add(
+ "FLINK_MOR_APPEND_ONLY_UNSUPPORTED",
+ "The bounded append-only path remains COPY_ON_WRITE",
+ )
+ else:
+ 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",
+ (
+ "FLINK_PR4_WRITE_PATH_UNSUPPORTED"
+ if mutable_mor_contract
+ else "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",
+ (
+ "FLINK_PR4_WRITE_PATH_UNSUPPORTED"
+ if mutable_mor_contract
+ else "FLINK_PR3_WRITE_PATH_UNSUPPORTED"
+ ),
"write.changelog_enabled must be explicitly false",
)
index = _dict(write.get("index"), "write.index", assessment)
@@ -932,6 +1004,22 @@ def assess_design(
"FLINK_MUTABLE_INDEX_BOOTSTRAP_REQUIRED",
"FLINK_STATE must bootstrap existing keys after state loss or
cold restart",
)
+ if mutable_mor_contract:
+ clustering = _dict(write.get("clustering"), "write.clustering",
assessment)
+ _check_keys(
+ clustering,
+ {"async_enabled", "schedule_enabled"},
+ "write.clustering",
+ assessment,
+ )
+ if (
+ clustering.get("schedule_enabled") is not False
+ or clustering.get("async_enabled") is not False
+ ):
+ assessment.add(
+ "FLINK_MOR_CLUSTERING_UNSUPPORTED",
+ "The bounded MOR path requires clustering schedule and
execution disabled",
+ )
else:
insert_cluster = write.get("insert_cluster")
if insert_cluster is True:
@@ -969,21 +1057,45 @@ 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"
+ expected_table_type = "MERGE_ON_READ" if mutable_mor_contract else
"COPY_ON_WRITE"
if (
"table.type" in connector_options
- and raw_table_type != "COPY_ON_WRITE"
+ and raw_table_type != expected_table_type
) or (
"write.operation" in connector_options
and raw_operation != expected_operation
):
assessment.add(
(
- "FLINK_PR3_WRITE_PATH_UNSUPPORTED"
- if mutable_contract
- else "FLINK_PR2_WRITE_PATH_UNSUPPORTED"
+ "FLINK_PR4_WRITE_PATH_UNSUPPORTED"
+ if mutable_mor_contract
+ else (
+ "FLINK_PR3_WRITE_PATH_UNSUPPORTED"
+ if mutable_contract
+ else "FLINK_PR2_WRITE_PATH_UNSUPPORTED"
+ )
),
- "A connector-option override leaves the bounded COW path",
+ "A connector-option override leaves the bounded write path",
)
+ if mutable_mor_contract:
+ if any(key.startswith("clustering.") for key in connector_options):
+ assessment.add(
+ "FLINK_MOR_CLUSTERING_UNSUPPORTED",
+ "Clustering connector overrides are not supported by the
bounded MOR path",
+ )
+ if any(
+ key in {
+ "compaction.tasks",
+ "compaction.delta_seconds",
+ "compaction.target.io",
+ "compaction.timeout.seconds",
+ }
+ for key in connector_options
+ ):
+ assessment.add(
+ "FLINK_MOR_COMPACTION_TUNING_UNSUPPORTED",
+ "Detailed compaction tuning is outside the bounded
Architect path",
+ )
verified_overrides = sorted(set(connector_options) -
set(unknown_options))
if verified_overrides:
assessment.add(
@@ -992,6 +1104,114 @@ def assess_design(
f"pass-through overrides are not rendered:
{verified_overrides}",
)
+ if mutable_mor_contract:
+ table_services = _dict(
+ contract.get("table_services"), "table_services", assessment
+ )
+ _check_keys(table_services, {"compaction"}, "table_services",
assessment)
+ compaction = _dict(
+ table_services.get("compaction"),
+ "table_services.compaction",
+ assessment,
+ )
+ compaction_keys = {
+ "async_enabled",
+ "delta_commits",
+ "mode",
+ "operation_execute_async_enabled",
+ "owner",
+ "schedule_enabled",
+ "trigger_strategy",
+ }
+ tuning_keys = {
+ "delta_seconds",
+ "target_io_mb",
+ "tasks",
+ "timeout_seconds",
+ }
+ unknown_compaction_keys = set(compaction) - compaction_keys
+ if unknown_compaction_keys & tuning_keys:
+ assessment.add(
+ "FLINK_MOR_COMPACTION_TUNING_UNSUPPORTED",
+ "Detailed compaction tuning is outside the bounded Architect
path: "
+ f"{sorted(unknown_compaction_keys & tuning_keys)}",
+ )
+ remaining_unknown = unknown_compaction_keys - tuning_keys
+ if remaining_unknown:
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "table_services.compaction contains unsupported fields: "
+ f"{sorted(remaining_unknown)}",
+ )
+ owner = compaction.get("owner")
+ if owner is None:
+ assessment.add(
+ "FLINK_MOR_COMPACTION_OWNER_REQUIRED",
+ "The MOR compaction owner must be explicit",
+ )
+ elif owner == "standalone_service":
+ assessment.add(
+ "FLINK_MULTI_WRITER_REVIEW",
+ "A standalone compactor is an independent table-service
writer",
+ )
+ elif owner != "flink_writer":
+ assessment.add(
+ "FLINK_WRITER_MODEL_UNRESOLVED",
+ "The MOR compaction owner is unresolved",
+ )
+ expected_compaction = {
+ "async_enabled": True,
+ "delta_commits": 5,
+ "mode": "async",
+ "operation_execute_async_enabled": True,
+ "schedule_enabled": True,
+ "trigger_strategy": "num_commits",
+ }
+ mismatches = [
+ key
+ for key, expected_value in expected_compaction.items()
+ if compaction.get(key) != expected_value
+ ]
+ if mismatches:
+ assessment.add(
+ "FLINK_MOR_COMPACTION_CONFIGURATION_UNSUPPORTED",
+ "The bounded MOR compaction contract has incompatible fields: "
+ f"{mismatches}",
+ )
+
+ read = _dict(contract.get("read"), "read", assessment)
+ _check_keys(
+ read,
+ {"query_modes", "read_optimized_freshness"},
+ "read",
+ assessment,
+ )
+ query_modes = _string_list(read.get("query_modes"),
"read.query_modes", assessment)
+ allowed_query_modes = {"snapshot", "read_optimized"}
+ if not query_modes or not
set(query_modes).issubset(allowed_query_modes):
+ assessment.add(
+ "FLINK_MOR_QUERY_REQUIREMENT_UNRESOLVED",
+ "MOR readers must explicitly require snapshot and/or
read_optimized queries",
+ )
+ read_optimized_freshness = read.get("read_optimized_freshness")
+ if "read_optimized" in query_modes:
+ if read_optimized_freshness is None or read_optimized_freshness ==
"unknown":
+ assessment.add(
+ "FLINK_MOR_QUERY_REQUIREMENT_UNRESOLVED",
+ "The acceptable read-optimized freshness must be explicit",
+ )
+ elif read_optimized_freshness != "after_compaction":
+ assessment.add(
+ "FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED",
+ "Read-optimized results can be current only after
compaction",
+ )
+ elif read_optimized_freshness != "NOT_APPLICABLE":
+ assessment.add(
+ "FLINK_DESIGN_CONTRACT_INVALID",
+ "read.read_optimized_freshness must be NOT_APPLICABLE without "
+ "read_optimized queries",
+ )
+
runtime = _dict(contract.get("runtime"), "runtime", assessment)
_check_keys(
runtime,
@@ -1012,9 +1232,13 @@ def assess_design(
elif execution_mode != "STREAMING":
assessment.add(
(
- "FLINK_PR3_WRITE_PATH_UNSUPPORTED"
- if mutable_contract
- else "FLINK_PR2_WRITE_PATH_UNSUPPORTED"
+ "FLINK_PR4_WRITE_PATH_UNSUPPORTED"
+ if mutable_mor_contract
+ else (
+ "FLINK_PR3_WRITE_PATH_UNSUPPORTED"
+ if mutable_contract
+ else "FLINK_PR2_WRITE_PATH_UNSUPPORTED"
+ )
),
"The executable path supports streaming execution",
)
diff --git
a/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr4/mutable_mor.json
b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr4/mutable_mor.json
new file mode 100644
index 000000000000..6f655b40eac2
--- /dev/null
+++
b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr4/mutable_mor.json
@@ -0,0 +1,85 @@
+{
+ "contract_schema": 3,
+ "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_mor_hudi",
+ "path": "file:///tmp/hudi-architect-mutable-mor-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": "MERGE_ON_READ",
+ "operation": "upsert",
+ "record_merge_mode": "EVENT_TIME_ORDERING",
+ "index": {
+ "type": "FLINK_STATE",
+ "global": true,
+ "state_ttl_days": 0,
+ "bootstrap_enabled": true
+ },
+ "changelog_enabled": false,
+ "clustering": {
+ "schedule_enabled": false,
+ "async_enabled": false
+ },
+ "connector_options": {}
+ },
+ "table_services": {
+ "compaction": {
+ "owner": "flink_writer",
+ "mode": "async",
+ "schedule_enabled": true,
+ "async_enabled": true,
+ "operation_execute_async_enabled": true,
+ "trigger_strategy": "num_commits",
+ "delta_commits": 5
+ }
+ },
+ "read": {
+ "query_modes": ["snapshot", "read_optimized"],
+ "read_optimized_freshness": "after_compaction"
+ },
+ "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_pr4/mutable_mor.sql
b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr4/mutable_mor.sql
new file mode 100644
index 000000000000..93d427b5ccb2
--- /dev/null
+++ b/hudi-agent-gateway/tests/fixtures/hudi_architect/flink_pr4/mutable_mor.sql
@@ -0,0 +1,62 @@
+/*
+ * 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_mor_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-mor-orders',
+ 'table.type' = 'MERGE_ON_READ',
+ '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',
+ 'compaction.schedule.enabled' = 'true',
+ 'compaction.async.enabled' = 'true',
+ 'compaction.operation.execute.async.enabled' = 'true',
+ 'compaction.trigger.strategy' = 'num_commits',
+ 'compaction.delta_commits' = '5',
+ 'clustering.schedule.enabled' = 'false',
+ 'clustering.async.enabled' = 'false'
+);
+
+INSERT INTO `orders_mutable_mor_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 b662da9e4a4d..39d885612bc2 100644
--- a/hudi-agent-gateway/tests/test_hudi_architect_skill.py
+++ b/hudi-agent-gateway/tests/test_hudi_architect_skill.py
@@ -41,6 +41,7 @@ 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"
+PR4_FIXTURE_DIR = GATEWAY_DIR / "tests" / "fixtures" / "hudi_architect" /
"flink_pr4"
ASF_LICENSE_MARKER = "Licensed to the Apache Software Foundation (ASF) under
one"
PR2_JAVA_FIXTURE = (
GATEWAY_DIR
@@ -170,7 +171,7 @@ def
test_flink_baseline_has_bounded_append_and_mutable_contracts() -> None:
"fixture_version": "1.20.1",
}
assert PINNED_HUDI_REVISION in capability
- assert manifest["schema_version"] == 3
+ assert manifest["schema_version"] == 4
assert manifest["fixture_artifacts"] == {
"hudi_bundle": "org.apache.hudi:hudi-flink1.20-bundle:1.2.0",
"flink_version": "1.20.1",
@@ -215,11 +216,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": 3,
+ "capability_manifest_schema": 4,
"design_contract_schema": 1,
"flink_fixture_version": "1.20.1",
"hudi_source_revision": PINNED_HUDI_REVISION,
- "supported_design_contract_schemas": [1, 2],
+ "supported_design_contract_schemas": [1, 2, 3],
}
@@ -348,6 +349,23 @@ def
test_flink_capability_manifest_records_executable_acceptance_contracts() ->
"table.type": "COPY_ON_WRITE",
"write.operation": "upsert",
},
+ "pr4-mutable-mor": {
+ "changelog.enabled": False,
+ "clustering.async.enabled": False,
+ "clustering.schedule.enabled": False,
+ "compaction.async.enabled": True,
+ "compaction.delta_commits": 5,
+ "compaction.operation.execute.async.enabled": True,
+ "compaction.schedule.enabled": True,
+ "compaction.trigger.strategy": "num_commits",
+ "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": "MERGE_ON_READ",
+ "write.operation": "upsert",
+ },
}
assert implemented_checks == {
"FLINK_APPEND_MODE_CLUSTERING_ENABLED": (
@@ -398,6 +416,26 @@ def
test_flink_capability_manifest_records_executable_acceptance_contracts() ->
"test_pr3_requires_index_bootstrap",
"testPinnedMutableIndexBootstrapConfiguration",
),
+ "FLINK_MOR_CLUSTERING_UNSUPPORTED": (
+ "test_pr4_rejects_clustering",
+ "testPinnedMorClusteringDisabled",
+ ),
+ "FLINK_MOR_APPEND_ONLY_UNSUPPORTED": (
+ "test_pr4_rejects_append_only_mor",
+ "testPinnedAppendOnlyMorDoesNotScheduleCompaction",
+ ),
+ "FLINK_MOR_COMPACTION_CONFIGURATION_UNSUPPORTED": (
+ "test_pr4_rejects_incompatible_compaction_configuration",
+ "testPinnedMorCompactionConfiguration",
+ ),
+ "FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED": (
+ "test_pr4_rejects_strict_read_optimized_freshness",
+ "testPinnedMorQuerySemantics",
+ ),
+ "FLINK_PR4_WRITE_PATH_UNSUPPORTED": (
+ "test_pr4_valid_mutable_mor_design_renders_exact_executable_sql",
+ "testMutableMorSinkAndCompactionPlan",
+ ),
}
for check in manifest["implemented_acceptance_checks"]:
assert check["python_test"] in python_tests
@@ -509,11 +547,11 @@ 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": 3,
+ "capability_manifest_schema": 4,
"design_contract_schema": 1,
"flink_fixture_version": "1.20.1",
"hudi_source_revision": PINNED_HUDI_REVISION,
- "supported_design_contract_schemas": [1, 2],
+ "supported_design_contract_schemas": [1, 2, 3],
}
@@ -530,7 +568,212 @@ def
test_pr3_valid_mutable_design_renders_exact_executable_sql() -> None:
PR3_FIXTURE_DIR / "mutable_cow.sql"
).rstrip("\n")
assert assessment["validation_evidence"]["design_contract_schema"] == 2
- assert assessment["validation_evidence"]["capability_manifest_schema"] == 3
+ assert assessment["validation_evidence"]["capability_manifest_schema"] == 4
+
+
+def test_pr4_valid_mutable_mor_design_renders_exact_executable_sql() -> None:
+ result = _run_flink_design_validator(PR4_FIXTURE_DIR / "mutable_mor.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(
+ PR4_FIXTURE_DIR / "mutable_mor.sql"
+ ).rstrip("\n")
+ assert assessment["validation_evidence"] == {
+ "baseline_id": "hudi-1.2.0-flink-1.20",
+ "capability_manifest_schema": 4,
+ "design_contract_schema": 3,
+ "flink_fixture_version": "1.20.1",
+ "hudi_source_revision": PINNED_HUDI_REVISION,
+ "supported_design_contract_schemas": [1, 2, 3],
+ }
+
+
[email protected](
+ ("field", "value"),
+ [
+ ("schedule_enabled", False),
+ ("async_enabled", False),
+ ("operation_execute_async_enabled", False),
+ ("mode", "inline"),
+ ("trigger_strategy", "time_elapsed"),
+ ("delta_commits", 6),
+ ],
+)
+def test_pr4_rejects_incompatible_compaction_configuration(
+ tmp_path: Path, field: str, value: object
+) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ design["table_services"]["compaction"][field] = value
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert "FLINK_MOR_COMPACTION_CONFIGURATION_UNSUPPORTED" in
assessment["finding_codes"]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
+
[email protected]("field", ["schedule_enabled", "async_enabled"])
+def test_pr4_rejects_clustering(tmp_path: Path, field: str) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ design["write"]["clustering"][field] = True
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == ["FLINK_MOR_CLUSTERING_UNSUPPORTED"]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
+
+def test_pr4_rejects_strict_read_optimized_freshness(tmp_path: Path) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ design["read"]["read_optimized_freshness"] = "after_every_commit"
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == [
+ "FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED"
+ ]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
+
[email protected]("freshness", [None, "unknown"])
+def test_pr4_requires_resolved_read_optimized_freshness(
+ tmp_path: Path, freshness: object
+) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ if freshness is None:
+ design["read"].pop("read_optimized_freshness")
+ else:
+ design["read"]["read_optimized_freshness"] = freshness
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == [
+ "FLINK_MOR_QUERY_REQUIREMENT_UNRESOLVED"
+ ]
+ assert assessment["status"] == "INCOMPLETE"
+ assert "artifacts" not in assessment
+
+
[email protected](
+ ("owner", "expected_code", "expected_status"),
+ [
+ (None, "FLINK_MOR_COMPACTION_OWNER_REQUIRED", "INCOMPLETE"),
+ ("standalone_service", "FLINK_MULTI_WRITER_REVIEW", "REVIEW_REQUIRED"),
+ ("unknown", "FLINK_WRITER_MODEL_UNRESOLVED", "REVIEW_REQUIRED"),
+ ],
+)
+def test_pr4_requires_same_job_compaction_owner(
+ tmp_path: Path, owner: object, expected_code: str, expected_status: str
+) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ if owner is None:
+ design["table_services"]["compaction"].pop("owner")
+ else:
+ design["table_services"]["compaction"]["owner"] = owner
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == [expected_code]
+ assert assessment["status"] == expected_status
+ assert "artifacts" not in assessment
+
+
+def test_pr4_rejects_detailed_compaction_tuning(tmp_path: Path) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ design["table_services"]["compaction"]["tasks"] = 4
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] ==
["FLINK_MOR_COMPACTION_TUNING_UNSUPPORTED"]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
+
+def test_pr4_preserves_combined_findings_and_withholds_artifacts(tmp_path:
Path) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ design["table_services"]["compaction"].pop("owner")
+ design["write"]["clustering"]["schedule_enabled"] = True
+ design["read"]["read_optimized_freshness"] = "after_every_commit"
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == [
+ "FLINK_MOR_CLUSTERING_UNSUPPORTED",
+ "FLINK_MOR_COMPACTION_OWNER_REQUIRED",
+ "FLINK_MOR_READ_OPTIMIZED_FRESHNESS_UNSUPPORTED",
+ ]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
+
+def test_pr4_snapshot_only_contract_does_not_invent_read_optimized_freshness(
+ tmp_path: Path,
+) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ design["read"] = {
+ "query_modes": ["snapshot"],
+ "read_optimized_freshness": "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"] == []
+
+
+def test_pr4_rejects_append_only_mor(tmp_path: Path) -> None:
+ design = json.loads(_read(PR2_FIXTURE_DIR / "stable_key.json"))
+ design["write"]["table_type"] = "MERGE_ON_READ"
+
+ result = _run_flink_design_validator(_write_design(tmp_path, design))
+
+ assert result.returncode == 1
+ assessment = json.loads(result.stdout)
+ assert assessment["finding_codes"] == ["FLINK_MOR_APPEND_ONLY_UNSUPPORTED"]
+ assert assessment["status"] == "BLOCKED"
+ assert "artifacts" not in assessment
+
+
[email protected](
+ ("field", "value"),
+ [("table_services", []), ("table_services", None), ("read", []), ("read",
None)],
+)
+def test_pr4_malformed_service_and_read_objects_fail_closed(
+ tmp_path: Path, field: str, value: object
+) -> None:
+ design = json.loads(_read(PR4_FIXTURE_DIR / "mutable_mor.json"))
+ design[field] = value
+
+ 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
@pytest.mark.parametrize(
@@ -734,7 +977,7 @@ 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])
[email protected]("contract_schema", [True, 1.0, 4, [], {}])
def test_flink_design_rejects_non_exact_contract_schema(
tmp_path: Path, contract_schema: object
) -> None:
@@ -745,6 +988,7 @@ def test_flink_design_rejects_non_exact_contract_schema(
assert result.returncode == 1
assessment = json.loads(result.stdout)
+ assert "Traceback" not in result.stderr
assert "FLINK_DESIGN_CONTRACT_INVALID" in assessment["finding_codes"]
assert assessment["executable_eligible"] is False
assert "artifacts" not in assessment
@@ -1316,7 +1560,8 @@ def test_safety_gate_order_is_stable() -> None:
"## F7 — Physical table and source contract",
"## F8 — Source changelog contract",
"## F9 — Streaming checkpoint contract",
- "## PR2/PR3 validation and completion",
+ "## F10 — Mutable table type, query freshness, and MOR ownership",
+ "## PR2/PR3/PR4 validation and completion",
]
offsets = [flow.index(heading) for heading in headings]
assert offsets == sorted(offsets)
