twalthr commented on code in PR #23601:
URL: https://github.com/apache/flink/pull/23601#discussion_r1372989815
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java:
##########
@@ -283,22 +293,29 @@ private abstract static class AbstractExactlyOnceSink
extends RichSinkFunction<R
private static final long serialVersionUID = 1L;
protected final String tableName;
+ protected final DataType consumedDataType;
+ protected final DataStructureConverter converter;
+ protected transient ListState<Row> rawResultState;
+ protected transient List<Row> localRawResult;
- protected transient ListState<String> rawResultState;
- protected transient List<String> localRawResult;
-
- protected AbstractExactlyOnceSink(String tableName) {
+ protected AbstractExactlyOnceSink(
+ String tableName, DataType consumedDataType,
DataStructureConverter converter) {
this.tableName = tableName;
+ this.consumedDataType = consumedDataType;
+ this.converter = converter;
}
@Override
public void initializeState(FunctionInitializationContext context)
throws Exception {
this.rawResultState =
context.getOperatorStateStore()
- .getListState(new
ListStateDescriptor<>("sink-results", Types.STRING));
+ .getListState(
+ new ListStateDescriptor<>(
+ "sink-results",
+
ExternalSerializer.of(consumedDataType, true)));
Review Comment:
Is the `true` correct here? The input is not internal but `Row`, no?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java:
##########
@@ -283,22 +293,29 @@ private abstract static class AbstractExactlyOnceSink
extends RichSinkFunction<R
private static final long serialVersionUID = 1L;
protected final String tableName;
+ protected final DataType consumedDataType;
+ protected final DataStructureConverter converter;
+ protected transient ListState<Row> rawResultState;
+ protected transient List<Row> localRawResult;
- protected transient ListState<String> rawResultState;
- protected transient List<String> localRawResult;
-
- protected AbstractExactlyOnceSink(String tableName) {
+ protected AbstractExactlyOnceSink(
+ String tableName, DataType consumedDataType,
DataStructureConverter converter) {
this.tableName = tableName;
+ this.consumedDataType = consumedDataType;
+ this.converter = converter;
}
@Override
public void initializeState(FunctionInitializationContext context)
throws Exception {
this.rawResultState =
context.getOperatorStateStore()
- .getListState(new
ListStateDescriptor<>("sink-results", Types.STRING));
+ .getListState(
+ new ListStateDescriptor<>(
+ "sink-results",
+
ExternalSerializer.of(consumedDataType, true)));
Review Comment:
if `localRawResult` would be RowData this would be true and we won't need
DataStructureConverter anymore
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
