You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ie...@apache.org on 2021/04/29 06:15:19 UTC
[beam] branch master updated (87e1164 -> 7418c84)
This is an automated email from the ASF dual-hosted git repository.
iemejia pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git.
from 87e1164 Merge pull request #14676: Revert "[BEAM-3713] Moving integration tests from nose to pytest (#14481)"
new b87f1f0 [BEAM-12247] Reduce memory/string creations in InMemoryTimerInternals
new be906f3 [BEAM-12248] Reduce ArrayList allocation in Row/RowUtils
new 7418c84 Merge pull request #14672: [BEAM-12248][BEAM-12247] Reduce String and ArrayList object creations
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../beam/runners/core/InMemoryTimerInternals.java | 46 ++++++++--------------
.../main/java/org/apache/beam/sdk/values/Row.java | 28 ++++++-------
.../java/org/apache/beam/sdk/values/RowUtils.java | 13 ++++++
3 files changed, 41 insertions(+), 46 deletions(-)
[beam] 01/03: [BEAM-12247] Reduce memory/string creations in
InMemoryTimerInternals
Posted by ie...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
iemejia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit b87f1f01197a78d60fb08a0ee7a4ef96bc4e08eb
Author: Daniel Kulp <dk...@apache.org>
AuthorDate: Wed Apr 28 14:16:31 2021 -0400
[BEAM-12247] Reduce memory/string creations in InMemoryTimerInternals
---
.../beam/runners/core/InMemoryTimerInternals.java | 46 ++++++++--------------
1 file changed, 16 insertions(+), 30 deletions(-)
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java
index 8be9081..d0b3bed 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/InMemoryTimerInternals.java
@@ -63,6 +63,9 @@ public class InMemoryTimerInternals implements TimerInternals {
/** Current synchronized processing time. */
private Instant synchronizedProcessingTime = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ /** Class.getSimpleName() cached to avoid allocations for tracing. */
+ private static final String SIMPLE_NAME = InMemoryTimerInternals.class.getSimpleName();
+
@Override
public @Nullable Instant currentOutputWatermarkTime() {
return outputWatermarkTime;
@@ -125,17 +128,12 @@ public class InMemoryTimerInternals implements TimerInternals {
@Deprecated
@Override
public void setTimer(TimerData timerData) {
- WindowTracing.trace("{}.setTimer: {}", getClass().getSimpleName(), timerData);
+ WindowTracing.trace("{}.setTimer: {}", SIMPLE_NAME, timerData);
- @Nullable
- TimerData existing =
- existingTimers.get(
- timerData.getNamespace(), timerData.getTimerId() + '+' + timerData.getTimerFamilyId());
+ @Nullable String colKey = timerData.getTimerId() + '+' + timerData.getTimerFamilyId();
+ TimerData existing = existingTimers.get(timerData.getNamespace(), colKey);
if (existing == null) {
- existingTimers.put(
- timerData.getNamespace(),
- timerData.getTimerId() + '+' + timerData.getTimerFamilyId(),
- timerData);
+ existingTimers.put(timerData.getNamespace(), colKey, timerData);
timersForDomain(timerData.getDomain()).add(timerData);
} else {
checkArgument(
@@ -149,10 +147,7 @@ public class InMemoryTimerInternals implements TimerInternals {
NavigableSet<TimerData> timers = timersForDomain(timerData.getDomain());
timers.remove(existing);
timers.add(timerData);
- existingTimers.put(
- timerData.getNamespace(),
- timerData.getTimerId() + '+' + timerData.getTimerFamilyId(),
- timerData);
+ existingTimers.put(timerData.getNamespace(), colKey, timerData);
}
}
}
@@ -216,7 +211,7 @@ public class InMemoryTimerInternals implements TimerInternals {
newInputWatermark);
WindowTracing.trace(
"{}.advanceInputWatermark: from {} to {}",
- getClass().getSimpleName(),
+ SIMPLE_NAME,
inputWatermarkTime,
newInputWatermark);
inputWatermarkTime = newInputWatermark;
@@ -229,7 +224,7 @@ public class InMemoryTimerInternals implements TimerInternals {
if (newOutputWatermark.isAfter(inputWatermarkTime)) {
WindowTracing.trace(
"{}.advanceOutputWatermark: clipping output watermark from {} to {}",
- getClass().getSimpleName(),
+ SIMPLE_NAME,
newOutputWatermark,
inputWatermarkTime);
adjustedOutputWatermark = inputWatermarkTime;
@@ -244,7 +239,7 @@ public class InMemoryTimerInternals implements TimerInternals {
adjustedOutputWatermark);
WindowTracing.trace(
"{}.advanceOutputWatermark: from {} to {}",
- getClass().getSimpleName(),
+ SIMPLE_NAME,
outputWatermarkTime,
adjustedOutputWatermark);
outputWatermarkTime = adjustedOutputWatermark;
@@ -259,10 +254,7 @@ public class InMemoryTimerInternals implements TimerInternals {
processingTime,
newProcessingTime);
WindowTracing.trace(
- "{}.advanceProcessingTime: from {} to {}",
- getClass().getSimpleName(),
- processingTime,
- newProcessingTime);
+ "{}.advanceProcessingTime: from {} to {}", SIMPLE_NAME, processingTime, newProcessingTime);
processingTime = newProcessingTime;
}
@@ -277,7 +269,7 @@ public class InMemoryTimerInternals implements TimerInternals {
newSynchronizedProcessingTime);
WindowTracing.trace(
"{}.advanceProcessingTime: from {} to {}",
- getClass().getSimpleName(),
+ SIMPLE_NAME,
synchronizedProcessingTime,
newSynchronizedProcessingTime);
synchronizedProcessingTime = newSynchronizedProcessingTime;
@@ -288,10 +280,7 @@ public class InMemoryTimerInternals implements TimerInternals {
TimerData timer = removeNextTimer(inputWatermarkTime, TimeDomain.EVENT_TIME);
if (timer != null) {
WindowTracing.trace(
- "{}.removeNextEventTimer: firing {} at {}",
- getClass().getSimpleName(),
- timer,
- inputWatermarkTime);
+ "{}.removeNextEventTimer: firing {} at {}", SIMPLE_NAME, timer, inputWatermarkTime);
}
return timer;
}
@@ -301,10 +290,7 @@ public class InMemoryTimerInternals implements TimerInternals {
TimerData timer = removeNextTimer(processingTime, TimeDomain.PROCESSING_TIME);
if (timer != null) {
WindowTracing.trace(
- "{}.removeNextProcessingTimer: firing {} at {}",
- getClass().getSimpleName(),
- timer,
- processingTime);
+ "{}.removeNextProcessingTimer: firing {} at {}", SIMPLE_NAME, timer, processingTime);
}
return timer;
}
@@ -316,7 +302,7 @@ public class InMemoryTimerInternals implements TimerInternals {
if (timer != null) {
WindowTracing.trace(
"{}.removeNextSynchronizedProcessingTimer: firing {} at {}",
- getClass().getSimpleName(),
+ SIMPLE_NAME,
timer,
synchronizedProcessingTime);
}
[beam] 02/03: [BEAM-12248] Reduce ArrayList allocation in
Row/RowUtils
Posted by ie...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
iemejia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit be906f38c5cac14c465332b8179022085ed76c13
Author: Daniel Kulp <dk...@apache.org>
AuthorDate: Wed Apr 28 14:21:13 2021 -0400
[BEAM-12248] Reduce ArrayList allocation in Row/RowUtils
---
.../main/java/org/apache/beam/sdk/values/Row.java | 28 ++++++++++------------
.../java/org/apache/beam/sdk/values/RowUtils.java | 13 ++++++++++
2 files changed, 25 insertions(+), 16 deletions(-)
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
index a19d9b3..9aa9277 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java
@@ -847,23 +847,19 @@ public abstract class Row implements Serializable {
+ " fields.");
}
- FieldOverrides fieldOverrides = new FieldOverrides(schema);
- fieldOverrides.setOverrides(this.values);
-
- Row row;
- if (!fieldOverrides.isEmpty()) {
- row =
- (Row)
- new RowFieldMatcher()
- .match(
- new CapturingRowCases(schema, fieldOverrides),
- FieldType.row(schema),
- new RowPosition(FieldAccessDescriptor.create()),
- null);
- } else {
- row = new RowWithStorage(schema, Collections.emptyList());
+ if (!values.isEmpty()) {
+ FieldOverrides fieldOverrides = new FieldOverrides(schema, this.values);
+ if (!fieldOverrides.isEmpty()) {
+ return (Row)
+ new RowFieldMatcher()
+ .match(
+ new CapturingRowCases(schema, fieldOverrides),
+ FieldType.row(schema),
+ new RowPosition(FieldAccessDescriptor.create()),
+ null);
+ }
}
- return row;
+ return new RowWithStorage(schema, Collections.emptyList());
}
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java
index d3b1c21..a5f6210 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowUtils.java
@@ -236,6 +236,11 @@ class RowUtils {
this.rootSchema = rootSchema;
}
+ FieldOverrides(Schema rootSchema, List<Object> overrides) {
+ this.topNode = new FieldAccessNode(rootSchema, overrides);
+ this.rootSchema = rootSchema;
+ }
+
boolean isEmpty() {
return topNode.isEmpty();
}
@@ -270,6 +275,14 @@ class RowUtils {
nestedAccess = Lists.newArrayList();
}
+ FieldAccessNode(Schema schema, List<Object> overrides) {
+ fieldOverrides = Lists.newArrayListWithExpectedSize(schema.getFieldCount());
+ for (Object value : overrides) {
+ fieldOverrides.add(new FieldOverride(value));
+ }
+ nestedAccess = Lists.newArrayList();
+ }
+
boolean isEmpty() {
return fieldOverrides.isEmpty() && nestedAccess.isEmpty();
}
[beam] 03/03: Merge pull request #14672: [BEAM-12248][BEAM-12247]
Reduce String and ArrayList object creations
Posted by ie...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
iemejia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 7418c84de2b68b06e0318766ca6c69f4d2144390
Merge: 87e1164 be906f3
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Thu Apr 29 08:13:58 2021 +0200
Merge pull request #14672: [BEAM-12248][BEAM-12247] Reduce String and ArrayList object creations
.../beam/runners/core/InMemoryTimerInternals.java | 46 ++++++++--------------
.../main/java/org/apache/beam/sdk/values/Row.java | 28 ++++++-------
.../java/org/apache/beam/sdk/values/RowUtils.java | 13 ++++++
3 files changed, 41 insertions(+), 46 deletions(-)