scwhittle commented on code in PR #39487: URL: https://github.com/apache/beam/pull/39487#discussion_r3676770885
########## runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/BatchDataflowOutputCounter.java: ########## @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.runners.dataflow.worker; + +import org.apache.beam.runners.core.ElementByteSizeObservable; +import org.apache.beam.runners.dataflow.worker.counters.CounterFactory; +import org.apache.beam.runners.dataflow.worker.counters.NameContext; + +/** + * A Dataflow output counter specific to Batch pipelines. In batch pipelines, empty-window elements + * (e.g. GroupingShuffleReader emitting KV<K, ValuesIterable>) represent a single PCollection + * element output. + */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) Review Comment: can this be removed? ########## runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java: ########## @@ -63,15 +84,17 @@ public void update(Object elem) throws Exception { objectAndByteCounter.update(elem); long windowsSize = ((WindowedValue<?>) elem).getWindows().size(); if (windowsSize == 0) { - // GroupingShuffleReader produces ValueInEmptyWindows. - // For now, we count the element at least once to keep the current counter - // behavior. - elementCount.addValue(1L); + updateEmptyWindows(elem); } else { + // Standard WindowedValue. elementCount.addValue(windowsSize); } } + protected void updateEmptyWindows(Object elem) throws Exception { Review Comment: have this take a WindowedValue<?> instead of Object ########## runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillKeyedWorkItem.java: ########## @@ -184,6 +184,56 @@ public Iterable<TimerData> timersIterable() { } } + @SuppressWarnings("nullness") + private @Nullable WindowedValue<ElemT> parseElemWindowOnly(Windmill.Message message) { Review Comment: can this share impl with parseElem? could delegate to internal method taking a bool on if the value should be parsed. Otherwise it might be easy to become inconsistent between the two ########## runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowOutputCounter.java: ########## @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.runners.dataflow.worker; + +import org.apache.beam.runners.core.ElementByteSizeObservable; +import org.apache.beam.runners.core.KeyedWorkItem; +import org.apache.beam.runners.dataflow.worker.counters.CounterFactory; +import org.apache.beam.runners.dataflow.worker.counters.NameContext; +import org.apache.beam.sdk.values.WindowedValue; + +/** + * A Dataflow output counter specific to Streaming pipelines. Unpacks {@link KeyedWorkItem}s in + * empty windows (e.g. emitted by WindowingWindmillReader) and counts element windows (ignoring + * timers) using lightweight metadata-only iteration via {@link + * KeyedWorkItem#elementWindowsIterable()}. + */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) Review Comment: ditto can this be removed? ########## runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowOutputCounter.java: ########## @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.runners.dataflow.worker; + +import org.apache.beam.runners.core.ElementByteSizeObservable; +import org.apache.beam.runners.core.KeyedWorkItem; +import org.apache.beam.runners.dataflow.worker.counters.CounterFactory; +import org.apache.beam.runners.dataflow.worker.counters.NameContext; +import org.apache.beam.sdk.values.WindowedValue; + +/** + * A Dataflow output counter specific to Streaming pipelines. Unpacks {@link KeyedWorkItem}s in + * empty windows (e.g. emitted by WindowingWindmillReader) and counts element windows (ignoring + * timers) using lightweight metadata-only iteration via {@link + * KeyedWorkItem#elementWindowsIterable()}. + */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) +}) +public class StreamingDataflowOutputCounter extends DataflowOutputCounter { + + public StreamingDataflowOutputCounter( + String outputName, CounterFactory counterFactory, NameContext nameContext) { + super(outputName, counterFactory, nameContext); + } + + public StreamingDataflowOutputCounter( + String outputName, + ElementByteSizeObservable<?> elementByteSizeObservable, + CounterFactory counterFactory, + NameContext nameContext) { + super(outputName, elementByteSizeObservable, counterFactory, nameContext); + } + + @Override + protected void updateEmptyWindows(Object elem) { Review Comment: if this is the only difference between the two, I might just prefer removing the heirarchy and just having a bool member variable on whether to introspect KeyedWorkItem values ########## runners/core-java/src/main/java/org/apache/beam/runners/core/KeyedWorkItem.java: ########## @@ -35,4 +35,13 @@ public interface KeyedWorkItem<K, ElemT> { /** Returns an iterable containing the elements. */ Iterable<WindowedValue<ElemT>> elementsIterable(); + + /** + * Returns an iterable containing windowed values without guaranteeing element payload decoding. + * Useful for lightweight inspection of windowing metadata without payload deserialization + * overhead. + */ + default Iterable<WindowedValue<ElemT>> elementWindowsIterable() { Review Comment: can this return WindowedValue<?> or WindowedValue<Void> so that callers know not to use the value? ########## runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java: ########## @@ -361,13 +361,24 @@ private Collection<W> windowsThatShouldFire(Set<W> windows) throws Exception { * setting holds, and invoking {@link ReduceFn#onTrigger}. * </ol> */ + public void processElements(KeyedWorkItem<?, InputT> keyedWorkItem) throws Exception { + processElementsInternal( + keyedWorkItem.elementWindowsIterable(), keyedWorkItem.elementsIterable()); + } + public void processElements(Iterable<WindowedValue<InputT>> values) throws Exception { - if (!values.iterator().hasNext()) { + processElementsInternal(values, values); + } + + private void processElementsInternal( + Iterable<WindowedValue<InputT>> elementWindows, Iterable<WindowedValue<InputT>> values) + throws Exception { + if (!elementWindows.iterator().hasNext()) { Review Comment: nit: use Iterables.empty() -- 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]
