[
https://issues.apache.org/jira/browse/FLINK-41274?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18137245#comment-18137245
]
Chanhae Oh edited comment on FLINK-41274 at 11/3/26, 2:26 PM:
--------------------------------------------------------------
Hi, [~twalthr] , [~panyuepeng].
Thanks for adding the testing instructions. I verified broadcast tables,
StateKind.BROADCAST, and ValueView for FLIP-565 on Flink 2.4 (built from
master), using a MiniClusterWithClientResource-backed JUnit 5 test. All
scenarios passed.
{code:java}
// PTF
@DataTypeHint("ROW<account STRING, total DOUBLE>")
public class RuleAggregatePtf extends ProcessTableFunction<Row> {
public void eval(
@StateHint ValueView<Double> total,
@StateHint(StateKind.BROADCAST) MapView<String, Double> multipliers,
@ArgumentHint(ArgumentTrait.SET_SEMANTIC_TABLE) Row data,
@ArgumentHint(ArgumentTrait.BROADCAST_SEMANTIC_TABLE) Row rule)
throws Exception {
if (rule != null) {
// Broadcast phase: write access to broadcast state only, no key
context available.
final String account = rule.getFieldAs("account");
final double multiplier = rule.getFieldAs("multiplier");
multipliers.put(account, multiplier);
return;
} final String account = data.getFieldAs("account");
final double amount = data.<Number>getFieldAs("amount").doubleValue();
final Double multiplier = multipliers.get(account);
final double weighted = amount * (multiplier == null ? 1.0 :
multiplier);
final double newTotal = (total.getValue() == null ? 0.0 :
total.getValue()) + weighted;
total.setValue(newTotal);
collect(Row.of(account, newTotal));
}
}{code}
Setup — input tables are fed via the DataStream API and registered as temporary
views; only the PTF call itself is SQL:
{code:java}
// tx: transaction rows (account STRING, amount DOUBLE)
// set-semantic table argument
DataStream<Row> txStream = env.fromSource(new FixedRowsSource(txRows), ...,
"transactions", ...);
tEnv.createTemporaryView("tx", txStream);
// rules: per-account multiplier rules (account STRING, multiplier DOUBLE)
// broadcast table argument
DataStream<Row> ruleStream = env.fromSource(new FixedRowsSource(ruleRows), ...,
"rules", ...);
tEnv.createTemporaryView("rules", ruleStream);
tEnv.createTemporaryFunction("RuleAggregate", RuleAggregatePtf.class);
Table resultTable = tEnv.sqlQuery(
"SELECT * FROM RuleAggregate(" +
" data => TABLE tx PARTITION BY account, " +
" rule => TABLE rules, " +
" uid => 'rule-aggregate-verify')");
tEnv.toDataStream(resultTable, Row.class).sinkTo(new
CollectingSink(sinkName));{code}
*Create a Flink SQL job that uses a PTF with broadcast tables, state and
ValueView. E.g. implementing an rule based aggregation* ✅
- Run 1 (parallelism 2): broadcast multiplier rules only — alice → 2.0, bob →
3.0
*Insert some rows and stop the job with a savepoint* ✅
- stopWithSavepoint on run 1 → savepoint1
- Run 2 (parallelism 3, restored from savepoint1): fed transactions alice:100,
bob:50, alice:20
- stopWithSavepoint on run 2 → savepoint2
*Resume from a savepoint (potentially also with a different parallelism) and
check that no state has been lost* ✅
- Run 3 (parallelism 4, restored from savepoint2): fed transactions alice:10,
bob:5, charlie:7 (charlie is a brand-new key, no rule ever broadcast for it)
*Raw test log output:*
{code:java}
Took savepoint 1 (rules broadcast only) at:
file:/.../junit-7077865994707775129/savepoint-20970d-d440cc0820da
22:59:31,708 INFO CheckpointCoordinator - Starting job a48e787a... from
savepoint
file:/.../savepoint-20970d-d440cc0820da (allowing non restored
state)
22:59:31,711 INFO CheckpointCoordinator - Restoring job a48e787a... from
Savepoint 2 @ 0
located at file:/.../savepoint-20970d-d440cc0820da.
Results after run 2 (batch 1 transactions): [+I[bob, bob, 150.0], +I[alice,
alice, 200.0], +I[alice, alice, 240.0]]
Took savepoint 2 (after batch 1) at: file:/.../savepoint-a48e78-ba856a150ad3
22:59:31,973 INFO CheckpointCoordinator - Starting job 93d6bcdd... from
savepoint
file:/.../savepoint-a48e78-ba856a150ad3 (allowing non restored
state)
22:59:31,974 INFO CheckpointCoordinator - Restoring job 93d6bcdd... from
Savepoint 4 @ 0
located at file:/.../savepoint-a48e78-ba856a150ad3.
Results after run 3 (batch 2 transactions): [+I[bob, bob, 150.0], +I[alice,
alice, 200.0], +I[alice, alice, 240.0],
+I[alice, alice, 260.0], +I[bob,
bob, 165.0], +I[charlie, charlie, 7.0]]
[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 8.831 s
-- in ptfverify.RuleAggregateSavepointVerifyTest
[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
[INFO] BUILD SUCCESS
{code}
- alice: (100+20)*2 + 10*2 = 260
- bob: 50*3 + 5*3 = 165
- charlie: 7 (default multiplier 1.0)
Both the keyed ValueView state and the broadcast MapView state survived two
consecutive savepoint/restore cycles with changing parallelism (2 → 3 → 4),
with no state loss. Ran the test twice to confirm it's stable.
{*}Environment{*}: Flink 2.4-SNAPSHOT (built from master),
MiniClusterWithClientResource (2 TaskManagers, 2 slots each).
was (Author: JIRAUSER311960):
Hi, [~twalthr] , [~panyuepeng].
Thanks for adding the testing instructions. I verified broadcast tables,
StateKind.BROADCAST, and ValueView for FLIP-565 on Flink 2.4 (built from
master), using a MiniClusterWithClientResource-backed JUnit 5 test. All
scenarios passed.
> Release Testing Verify: FLIP-565: Improve ProcessTableFunctions for late data
> handling and state access
> -------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41274
> URL: https://issues.apache.org/jira/browse/FLINK-41274
> Project: Flink
> Issue Type: Sub-task
> Components: Tests
> Affects Versions: 2.4.0
> Reporter: Yuepeng Pan
> Assignee: Chanhae Oh
> Priority: Blocker
> Fix For: 2.4.0
>
>
> * Create a Flink SQL job that uses a PTF with broadcast tables, state and
> ValueView. E.g. implementing an rule based aggregation
> * Insert some rows and stop the job with a savepoint
> * Resume from a savepoint (potentially also with a different parallelism)
> and check that no state has been lost.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
