SEPURI-SAI-KRISHNA commented on code in PR #29345:
URL: https://github.com/apache/flink/pull/29345#discussion_r4284094426
##########
flink-table/flink-table-test-utils/src/main/java/org/apache/flink/table/runtime/functions/ProcessTableFunctionTestHarness.java:
##########
@@ -234,6 +239,58 @@ public static <OUT> Builder<OUT> ofClass(
return new Builder<>(functionClass);
}
+ //
-------------------------------------------------------------------------
+ // Test State Snapshotting
+ //
-------------------------------------------------------------------------
+
+ /**
+ * Takes a snapshot of this harness: the configuration it was built with,
the state of every
+ * partition, the pending and the fired timers, the watermark of every
table argument, and the
+ * collected output.
+ *
+ * <p>The snapshot is isolated from this harness, so processing further
elements, advancing a
+ * watermark or mutating state afterwards does not change what the
snapshot holds. State is
+ * copied in its internal representation, which means no user-facing state
object is touched.
+ * The one part that is not copied is the values collected by the PTF
itself (see {@link
+ * #getFunctionOutput()}), since those are the instances the function
emitted.
+ *
+ * @see #restoreFromSnapshot(TestSnapshot)
+ */
+ public TestSnapshot<OUT> snapshot() {
+ return new TestSnapshot<>(
+ builder,
+ stateManager.snapshotState(),
+ timerManager.snapshot(),
+ new ArrayList<>(functionOutput),
+ copyRows(output));
+ }
+
+ /**
+ * Creates a new harness that continues from the given snapshot.
+ *
+ * <p>The returned harness runs a fresh instance of the PTF, opened as
{@link Builder#build()}
+ * would, and starts out with the state, timers, watermarks and output
that the snapshot was
+ * taken with. A snapshot can be restored more than once, and each
restored harness is
+ * independent of the others.
+ *
+ * @see #snapshot()
+ */
+ public static <OUT> ProcessTableFunctionTestHarness<OUT>
restoreFromSnapshot(
+ TestSnapshot<OUT> snapshot) throws Exception {
+ checkNotNull(snapshot, "snapshot must not be null");
+
+ final ProcessTableFunctionTestHarness<OUT> harness =
snapshot.builder.build();
+ harness.stateManager.restoreState(snapshot.state);
+ harness.timerManager.restore(snapshot.timers);
+ harness.functionOutput.addAll(snapshot.functionOutput);
+ harness.output.addAll(copyRows(snapshot.output));
+ return harness;
+ }
+
+ private static List<Row> copyRows(List<Row> rows) {
+ return rows.stream().map(Row::copy).collect(Collectors.toList());
Review Comment:
Fixed, and your snippet is the test:
`testSnapshotIsNotAffectedByMutatingANestedOutputValue`. It applied to three
more places I had written: the state map keys, the pending timer map keys and
`Timer`'s own key. Those are worse, because `Row` hashes deeply, so a shared
array inside a key changes that key's hash and can make the snapshot's own
entry unreachable. `PARTITION BY` on an `ARRAY<INT>` column is accepted, so it
is reachable. Everything deep copies through `ExternalSerializer` now, as you
suggested.
--
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]
