cshuo commented on code in PR #20281:
URL: https://github.com/apache/hudi/pull/20281#discussion_r4245364515
##########
hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py:
##########
@@ -651,8 +733,48 @@ 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",
+ )
+ for field in ordering_fields:
Review Comment:
Reject `no_precombine` as an ordering field
A non-null BIGINT column named `no_precombine` passes validation and
produces `'ordering.fields' = 'no_precombine'`. Hudi 1.2.0 treats this value as
a sentinel and resolves the ordering fields to null, rather than using the
column. Please reject this reserved value before emitting SQL.
##########
hudi-agent-gateway/skills/hudi-architect/validate_flink_design.py:
##########
@@ -491,20 +542,34 @@ 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 (
+ {"must_collapse"} if mutable_contract else {"cannot_occur",
"duplicates_acceptable"}
Review Comment:
Accept resolved replay answers for mutable contracts
F6 says both “replays cannot occur” and “duplicates are acceptable” can
continue, but schema 2 accepts only `must_collapse`. An otherwise valid mutable
contract with either answer returns `FLINK_REPLAY_BEHAVIOR_UNRESOLVED` and
emits no SQL. Please align the accepted values with F6.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
