bvarghese1 commented on code in PR #26424:
URL: https://github.com/apache/flink/pull/26424#discussion_r2182521258
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/NonTimeRangeUnboundedPrecedingFunction.java:
##########
@@ -168,127 +91,19 @@ public NonTimeRangeUnboundedPrecedingFunction(
LogicalType[] accTypes,
LogicalType[] inputFieldTypes,
LogicalType[] sortKeyTypes,
- RowDataKeySelector sortKeySelector) {
- this.stateRetentionTime = stateRetentionTime;
- this.generatedAggsHandler = genAggsHandler;
- this.generatedRecordEqualiser = genRecordEqualiser;
- this.generatedSortKeyEqualiser = genSortKeyEqualiser;
- this.generatedSortKeyComparator = genSortKeyComparator;
- this.accTypes = accTypes;
- this.inputFieldTypes = inputFieldTypes;
- this.sortKeyTypes = sortKeyTypes;
- this.sortKeySelector = sortKeySelector;
- }
-
- @Override
- public void open(OpenContext openContext) throws Exception {
- // Initialize agg functions
- aggFuncs =
generatedAggsHandler.newInstance(getRuntimeContext().getUserCodeClassLoader());
- aggFuncs.open(new PerKeyStateDataViewStore(getRuntimeContext()));
-
- // Initialize output record
- output = new JoinedRowData();
-
- // Initialize value/row equaliser
- valueEqualiser =
-
generatedRecordEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());
-
- // Initialize sortKey equaliser
- sortKeyEqualiser =
-
generatedSortKeyEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());
-
- // Initialize sort comparator
- sortKeyComparator =
- generatedSortKeyComparator.newInstance(
- getRuntimeContext().getUserCodeClassLoader());
-
- StateTtlConfig ttlConfig = createTtlConfig(stateRetentionTime);
-
- // Initialize state to maintain id counter
- idStateDescriptor = new ValueStateDescriptor<Long>("idState",
Long.class);
- if (ttlConfig.isEnabled()) {
- idStateDescriptor.enableTimeToLive(ttlConfig);
- }
- idState = getRuntimeContext().getState(idStateDescriptor);
-
- // Input elements are all binary rows as they came from network
- InternalTypeInfo<RowData> inputRowTypeInfo =
InternalTypeInfo.ofFields(inputFieldTypes);
- InternalTypeInfo<RowData> sortKeyRowTypeInfo =
InternalTypeInfo.ofFields(this.sortKeyTypes);
-
- // Initialize state which maintains a sorted list of tuples(sortKey,
List of IDs)
- ListTypeInfo<Long> idListTypeInfo = new ListTypeInfo<Long>(Types.LONG);
- ListTypeInfo<Tuple2<RowData, List<Long>>> listTypeInfo =
- new ListTypeInfo<>(new TupleTypeInfo<>(sortKeyRowTypeInfo,
idListTypeInfo));
- sortedListStateDescriptor =
- new ValueStateDescriptor<List<Tuple2<RowData, List<Long>>>>(
- "sortedListState", listTypeInfo);
- if (ttlConfig.isEnabled()) {
- sortedListStateDescriptor.enableTimeToLive(ttlConfig);
- }
- sortedListState =
getRuntimeContext().getState(sortedListStateDescriptor);
-
- // Initialize state which maintains the actual row
- valueStateDescriptor =
- new MapStateDescriptor<Long, RowData>(
- "valueMapState", Types.LONG, inputRowTypeInfo);
- if (ttlConfig.isEnabled()) {
- valueStateDescriptor.enableTimeToLive(ttlConfig);
- }
- valueMapState = getRuntimeContext().getMapState(valueStateDescriptor);
-
- // Initialize accumulator state per row
- InternalTypeInfo<RowData> accTypeInfo =
InternalTypeInfo.ofFields(accTypes);
- accStateDescriptor =
- new MapStateDescriptor<RowData, RowData>(
- "accMapState", sortKeyRowTypeInfo, accTypeInfo);
- if (ttlConfig.isEnabled()) {
- accStateDescriptor.enableTimeToLive(ttlConfig);
- }
- accMapState = getRuntimeContext().getMapState(accStateDescriptor);
-
- // metrics
- this.numOfIdsNotFound =
-
getRuntimeContext().getMetricGroup().counter(IDS_NOT_FOUND_METRIC_NAME);
- this.numOfSortKeysNotFound =
-
getRuntimeContext().getMetricGroup().counter(SORT_KEYS_NOT_FOUND_METRIC_NAME);
- }
-
- /**
- * Puts an element from the input stream into state or removes it from
state if the input is a
- * retraction. Emits the aggregated value for the newly inserted element
and updates all results
- * that are affected by the added or removed row. Emits the same
aggregated value for all
- * elements with the same sortKey to comply with the sql RANGE syntax.
- *
- * @param input The input value.
- * @param ctx A {@link Context} that allows querying the timestamp of the
element and getting
- * TimerService for registering timers and querying the time. The
context is only valid
- * during the invocation of this method, do not store it.
- * @param out The collector for returning result values.
- * @throws Exception
- */
- @Override
- public void processElement(
- RowData input,
- KeyedProcessFunction<K, RowData, RowData>.Context ctx,
- Collector<RowData> out)
- throws Exception {
- RowKind rowKind = input.getRowKind();
-
- switch (rowKind) {
- case INSERT:
- case UPDATE_AFTER:
- insertIntoSortedList(input, out);
- break;
-
- case DELETE:
- case UPDATE_BEFORE:
- removeFromSortedList(input, out);
- break;
- }
-
- // Reset acc state since we can have out of order inserts into the
ordered list
- aggFuncs.resetAccumulators();
- aggFuncs.cleanup();
+ RowDataKeySelector sortKeySelector,
+ InternalTypeInfo<RowData> accKeyRowTypeInfo) {
Review Comment:
Removed the last parameter `InternalTypeInfo<RowData> accKeyRowTypeInfo`
--
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
