You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2016/03/24 03:47:31 UTC
[07/67] [partial] incubator-beam git commit: Directory reorganization
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UnownedInputStream.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UnownedInputStream.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UnownedInputStream.java
deleted file mode 100644
index 3d80230..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UnownedInputStream.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import com.google.common.base.MoreObjects;
-
-import java.io.FilterInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-
-/**
- * A {@link OutputStream} wrapper which protects against the user attempting to modify
- * the underlying stream by closing it or using mark.
- */
-public class UnownedInputStream extends FilterInputStream {
- public UnownedInputStream(InputStream delegate) {
- super(delegate);
- }
-
- @Override
- public void close() throws IOException {
- throw new UnsupportedOperationException("Caller does not own the underlying input stream "
- + " and should not call close().");
- }
-
- @Override
- public boolean equals(Object obj) {
- return obj instanceof UnownedInputStream
- && ((UnownedInputStream) obj).in.equals(in);
- }
-
- @Override
- public int hashCode() {
- return in.hashCode();
- }
-
- @SuppressWarnings("UnsynchronizedOverridesSynchronized")
- @Override
- public void mark(int readlimit) {
- throw new UnsupportedOperationException("Caller does not own the underlying input stream "
- + " and should not call mark().");
- }
-
- @Override
- public boolean markSupported() {
- return false;
- }
-
- @SuppressWarnings("UnsynchronizedOverridesSynchronized")
- @Override
- public void reset() throws IOException {
- throw new UnsupportedOperationException("Caller does not own the underlying input stream "
- + " and should not call reset().");
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(UnownedInputStream.class).add("in", in).toString();
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UnownedOutputStream.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UnownedOutputStream.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UnownedOutputStream.java
deleted file mode 100644
index 29187a1..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UnownedOutputStream.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import com.google.common.base.MoreObjects;
-
-import java.io.FilterOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-
-/**
- * A {@link OutputStream} wrapper which protects against the user attempting to modify
- * the underlying stream by closing it.
- */
-public class UnownedOutputStream extends FilterOutputStream {
- public UnownedOutputStream(OutputStream delegate) {
- super(delegate);
- }
-
- @Override
- public void close() throws IOException {
- throw new UnsupportedOperationException("Caller does not own the underlying output stream "
- + " and should not call close().");
- }
-
- @Override
- public boolean equals(Object obj) {
- return obj instanceof UnownedOutputStream
- && ((UnownedOutputStream) obj).out.equals(out);
- }
-
- @Override
- public int hashCode() {
- return out.hashCode();
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(UnownedOutputStream.class).add("out", out).toString();
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UploadIdResponseInterceptor.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UploadIdResponseInterceptor.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UploadIdResponseInterceptor.java
deleted file mode 100644
index da597e6..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UploadIdResponseInterceptor.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import com.google.api.client.http.GenericUrl;
-import com.google.api.client.http.HttpResponse;
-import com.google.api.client.http.HttpResponseInterceptor;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-
-/**
- * Implements a response intercepter that logs the upload id if the upload
- * id header exists and it is the first request (does not have upload_id parameter in the request).
- * Only logs if debug level is enabled.
- */
-public class UploadIdResponseInterceptor implements HttpResponseInterceptor {
-
- private static final Logger LOG = LoggerFactory.getLogger(UploadIdResponseInterceptor.class);
- private static final String UPLOAD_ID_PARAM = "upload_id";
- private static final String UPLOAD_TYPE_PARAM = "uploadType";
- private static final String UPLOAD_HEADER = "X-GUploader-UploadID";
-
- @Override
- public void interceptResponse(HttpResponse response) throws IOException {
- if (!LOG.isDebugEnabled()) {
- return;
- }
- String uploadId = response.getHeaders().getFirstHeaderStringValue(UPLOAD_HEADER);
- if (uploadId == null) {
- return;
- }
-
- GenericUrl url = response.getRequest().getUrl();
- // The check for no upload id limits the output to one log line per upload.
- // The check for upload type makes sure this is an upload and not a read.
- if (url.get(UPLOAD_ID_PARAM) == null && url.get(UPLOAD_TYPE_PARAM) != null) {
- LOG.debug(
- "Upload ID for url {} on worker {} is {}",
- url,
- System.getProperty("worker_id"),
- uploadId);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UserCodeException.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UserCodeException.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UserCodeException.java
deleted file mode 100644
index 9b9c7a5..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/UserCodeException.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import java.util.Arrays;
-import java.util.Objects;
-
-/**
- * An exception that was thrown in user-code. Sets the stack trace
- * from the first time execution enters user code down through the
- * rest of the user's stack frames until the exception is
- * reached.
- */
-public class UserCodeException extends RuntimeException {
-
- public static UserCodeException wrap(Throwable t) {
- if (t instanceof UserCodeException) {
- return (UserCodeException) t;
- }
-
- return new UserCodeException(t);
- }
-
- public static RuntimeException wrapIf(boolean condition, Throwable t) {
- if (condition) {
- return wrap(t);
- }
-
- if (t instanceof RuntimeException) {
- return (RuntimeException) t;
- }
-
- return new RuntimeException(t);
- }
-
- private UserCodeException(Throwable t) {
- super(t);
- truncateStackTrace(t);
- }
-
- /**
- * Truncates the @{Throwable}'s stack trace to contain only user code,
- * removing all frames below.
- *
- * <p>This is to remove infrastructure noise below user code entry point. We do this
- * by finding common stack frames between the throwable's captured stack and that
- * of the current thread.
- */
- private void truncateStackTrace(Throwable t) {
-
- StackTraceElement[] currentStack = Thread.currentThread().getStackTrace();
- StackTraceElement[] throwableStack = t.getStackTrace();
-
- int currentStackSize = currentStack.length;
- int throwableStackSize = throwableStack.length;
-
- int commonFrames = 0;
- while (framesEqual(currentStack[currentStackSize - commonFrames - 1],
- throwableStack[throwableStackSize - commonFrames - 1])) {
- commonFrames++;
- if (commonFrames >= Math.min(currentStackSize, throwableStackSize)) {
- break;
- }
- }
-
- StackTraceElement[] truncatedStack = Arrays.copyOfRange(throwableStack, 0,
- throwableStackSize - commonFrames);
- t.setStackTrace(truncatedStack);
- }
-
- /**
- * Check if two frames are equal; Frames are considered equal if they point to the same method.
- */
- private boolean framesEqual(StackTraceElement frame1, StackTraceElement frame2) {
- boolean areEqual = Objects.equals(frame1.getClassName(), frame2.getClassName());
- areEqual &= Objects.equals(frame1.getMethodName(), frame2.getMethodName());
-
- return areEqual;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/ValueWithRecordId.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/ValueWithRecordId.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/ValueWithRecordId.java
deleted file mode 100644
index ac1f2eb..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/ValueWithRecordId.java
+++ /dev/null
@@ -1,154 +0,0 @@
-/*******************************************************************************
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import com.google.cloud.dataflow.sdk.coders.ByteArrayCoder;
-import com.google.cloud.dataflow.sdk.coders.Coder;
-import com.google.cloud.dataflow.sdk.coders.StandardCoder;
-import com.google.cloud.dataflow.sdk.transforms.DoFn;
-import com.google.cloud.dataflow.sdk.transforms.PTransform;
-import com.google.cloud.dataflow.sdk.transforms.ParDo;
-import com.google.cloud.dataflow.sdk.values.PCollection;
-import com.google.common.base.MoreObjects;
-import com.google.common.base.Preconditions;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Objects;
-
-/**
- * Immutable struct containing a value as well as a unique id identifying the value.
- *
- * @param <ValueT> the underlying value type
- */
-public class ValueWithRecordId<ValueT> {
- private final ValueT value;
- private final byte[] id;
-
- public ValueWithRecordId(ValueT value, byte[] id) {
- this.value = value;
- this.id = id;
- }
-
- public ValueT getValue() {
- return value;
- }
-
- public byte[] getId() {
- return id;
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(this)
- .add("id", id)
- .add("value", value)
- .toString();
- }
-
- @Override
- public boolean equals(Object other) {
- if (this == other) {
- return true;
- }
- if (!(other instanceof ValueWithRecordId)) {
- return false;
- }
- ValueWithRecordId<?> otherRecord = (ValueWithRecordId<?>) other;
- return Objects.deepEquals(id, otherRecord.id)
- && Objects.deepEquals(value, otherRecord.value);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(Arrays.hashCode(id), value);
- }
-
- /**
- * A {@link Coder} for {@code ValueWithRecordId}, using a wrapped value {@code Coder}.
- */
- public static class ValueWithRecordIdCoder<ValueT>
- extends StandardCoder<ValueWithRecordId<ValueT>> {
- public static <ValueT> ValueWithRecordIdCoder<ValueT> of(Coder<ValueT> valueCoder) {
- return new ValueWithRecordIdCoder<>(valueCoder);
- }
-
- @JsonCreator
- public static <ValueT> ValueWithRecordIdCoder<ValueT> of(
- @JsonProperty(PropertyNames.COMPONENT_ENCODINGS)
- List<Coder<ValueT>> components) {
- Preconditions.checkArgument(components.size() == 1,
- "Expecting 1 component, got " + components.size());
- return of(components.get(0));
- }
-
- protected ValueWithRecordIdCoder(Coder<ValueT> valueCoder) {
- this.valueCoder = valueCoder;
- this.idCoder = ByteArrayCoder.of();
- }
-
- @Override
- public List<? extends Coder<?>> getCoderArguments() {
- return Arrays.asList(valueCoder);
- }
-
- @Override
- public void encode(ValueWithRecordId<ValueT> value, OutputStream outStream, Context context)
- throws IOException {
- valueCoder.encode(value.value, outStream, context.nested());
- idCoder.encode(value.id, outStream, context);
- }
-
- @Override
- public ValueWithRecordId<ValueT> decode(InputStream inStream, Context context)
- throws IOException {
- return new ValueWithRecordId<ValueT>(
- valueCoder.decode(inStream, context.nested()),
- idCoder.decode(inStream, context));
- }
-
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- valueCoder.verifyDeterministic();
- }
-
- public Coder<ValueT> getValueCoder() {
- return valueCoder;
- }
-
- Coder<ValueT> valueCoder;
- ByteArrayCoder idCoder;
- }
-
- public static <T>
- PTransform<PCollection<? extends ValueWithRecordId<T>>, PCollection<T>> stripIds() {
- return ParDo.named("StripIds")
- .of(
- new DoFn<ValueWithRecordId<T>, T>() {
- @Override
- public void processElement(ProcessContext c) {
- c.output(c.element().getValue());
- }
- });
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Values.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Values.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Values.java
deleted file mode 100644
index d4440e7..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Values.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import java.util.Map;
-
-import javax.annotation.Nullable;
-
-/**
- * A collection of static methods for manipulating value representations
- * transfered via the Dataflow API.
- */
-public final class Values {
- private Values() {} // Non-instantiable
-
- public static Boolean asBoolean(Object value) throws ClassCastException {
- @Nullable Boolean knownResult = checkKnownValue(CloudKnownType.BOOLEAN, value, Boolean.class);
- if (knownResult != null) {
- return knownResult;
- }
- return Boolean.class.cast(value);
- }
-
- public static Double asDouble(Object value) throws ClassCastException {
- @Nullable Double knownResult = checkKnownValue(CloudKnownType.FLOAT, value, Double.class);
- if (knownResult != null) {
- return knownResult;
- }
- if (value instanceof Double) {
- return (Double) value;
- }
- return ((Float) value).doubleValue();
- }
-
- public static Long asLong(Object value) throws ClassCastException {
- @Nullable Long knownResult = checkKnownValue(CloudKnownType.INTEGER, value, Long.class);
- if (knownResult != null) {
- return knownResult;
- }
- if (value instanceof Long) {
- return (Long) value;
- }
- return ((Integer) value).longValue();
- }
-
- public static String asString(Object value) throws ClassCastException {
- @Nullable String knownResult = checkKnownValue(CloudKnownType.TEXT, value, String.class);
- if (knownResult != null) {
- return knownResult;
- }
- return String.class.cast(value);
- }
-
- @Nullable
- private static <T> T checkKnownValue(CloudKnownType type, Object value, Class<T> clazz) {
- if (!(value instanceof Map)) {
- return null;
- }
- Map<String, Object> map = (Map<String, Object>) value;
- @Nullable String typeName = (String) map.get(PropertyNames.OBJECT_TYPE_NAME);
- if (typeName == null) {
- return null;
- }
- @Nullable CloudKnownType knownType = CloudKnownType.forUri(typeName);
- if (knownType == null || knownType != type) {
- return null;
- }
- @Nullable Object scalar = map.get(PropertyNames.SCALAR_FIELD_NAME);
- if (scalar == null) {
- return null;
- }
- return knownType.parse(scalar, clazz);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/VarInt.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/VarInt.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/VarInt.java
deleted file mode 100644
index af03911..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/VarInt.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-
-/**
- * Variable-length encoding for integers.
- *
- * <p>Handles, in a common encoding format, signed bytes, shorts, ints, and longs.
- * Takes between 1 and 10 bytes.
- * Less efficient than BigEndian{Int,Long} coder for negative or large numbers.
- * All negative ints are encoded using 5 bytes, longs take 10 bytes.
- */
-public class VarInt {
-
- private static long convertIntToLongNoSignExtend(int v) {
- return v & 0xFFFFFFFFL;
- }
-
- /**
- * Encodes the given value onto the stream.
- */
- public static void encode(int v, OutputStream stream) throws IOException {
- encode(convertIntToLongNoSignExtend(v), stream);
- }
-
- /**
- * Encodes the given value onto the stream.
- */
- public static void encode(long v, OutputStream stream) throws IOException {
- do {
- // Encode next 7 bits + terminator bit
- long bits = v & 0x7F;
- v >>>= 7;
- byte b = (byte) (bits | ((v != 0) ? 0x80 : 0));
- stream.write(b);
- } while (v != 0);
- }
-
- /**
- * Decodes an integer value from the given stream.
- */
- public static int decodeInt(InputStream stream) throws IOException {
- long r = decodeLong(stream);
- if (r < 0 || r >= 1L << 32) {
- throw new IOException("varint overflow " + r);
- }
- return (int) r;
- }
-
- /**
- * Decodes a long value from the given stream.
- */
- public static long decodeLong(InputStream stream) throws IOException {
- long result = 0;
- int shift = 0;
- int b;
- do {
- // Get 7 bits from next byte
- b = stream.read();
- if (b < 0) {
- if (shift == 0) {
- throw new EOFException();
- } else {
- throw new IOException("varint not terminated");
- }
- }
- long bits = b & 0x7F;
- if (shift >= 64 || (shift == 63 && bits > 1)) {
- // Out of range
- throw new IOException("varint too long");
- }
- result |= bits << shift;
- shift += 7;
- } while ((b & 0x80) != 0);
- return result;
- }
-
- /**
- * Returns the length of the encoding of the given value (in bytes).
- */
- public static int getLength(int v) {
- return getLength(convertIntToLongNoSignExtend(v));
- }
-
- /**
- * Returns the length of the encoding of the given value (in bytes).
- */
- public static int getLength(long v) {
- int result = 0;
- do {
- result++;
- v >>>= 7;
- } while (v != 0);
- return result;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WatermarkHold.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WatermarkHold.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WatermarkHold.java
deleted file mode 100644
index d537ddb..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WatermarkHold.java
+++ /dev/null
@@ -1,450 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import com.google.cloud.dataflow.sdk.transforms.windowing.BoundedWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.OutputTimeFn;
-import com.google.cloud.dataflow.sdk.transforms.windowing.OutputTimeFns;
-import com.google.cloud.dataflow.sdk.transforms.windowing.Window.ClosingBehavior;
-import com.google.cloud.dataflow.sdk.util.state.MergingStateAccessor;
-import com.google.cloud.dataflow.sdk.util.state.ReadableState;
-import com.google.cloud.dataflow.sdk.util.state.StateMerging;
-import com.google.cloud.dataflow.sdk.util.state.StateTag;
-import com.google.cloud.dataflow.sdk.util.state.StateTags;
-import com.google.cloud.dataflow.sdk.util.state.WatermarkHoldState;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-
-import org.joda.time.Duration;
-import org.joda.time.Instant;
-
-import java.io.Serializable;
-
-import javax.annotation.Nullable;
-
-/**
- * Implements the logic to hold the output watermark for a computation back
- * until it has seen all the elements it needs based on the input watermark for the
- * computation.
- *
- * <p>The backend ensures the output watermark can never progress beyond the
- * input watermark for a computation. GroupAlsoByWindows computations may add a 'hold'
- * to the output watermark in order to prevent it progressing beyond a time within a window.
- * The hold will be 'cleared' when the associated pane is emitted.
- *
- * <p>This class is only intended for use by {@link ReduceFnRunner}. The two evolve together and
- * will likely break any other uses.
- *
- * @param <W> The kind of {@link BoundedWindow} the hold is for.
- */
-class WatermarkHold<W extends BoundedWindow> implements Serializable {
- /**
- * Return tag for state containing the output watermark hold
- * used for elements.
- */
- public static <W extends BoundedWindow>
- StateTag<Object, WatermarkHoldState<W>> watermarkHoldTagForOutputTimeFn(
- OutputTimeFn<? super W> outputTimeFn) {
- return StateTags.<Object, WatermarkHoldState<W>>makeSystemTagInternal(
- StateTags.<W>watermarkStateInternal("hold", outputTimeFn));
- }
-
- /**
- * Tag for state containing end-of-window and garbage collection output watermark holds.
- * (We can't piggy-back on the data hold state since the outputTimeFn may be
- * {@link OutputTimeFns#outputAtLatestInputTimestamp()}, in which case every pane will
- * would take the end-of-window time as its element time.)
- */
- @VisibleForTesting
- public static final StateTag<Object, WatermarkHoldState<BoundedWindow>> EXTRA_HOLD_TAG =
- StateTags.makeSystemTagInternal(StateTags.watermarkStateInternal(
- "extra", OutputTimeFns.outputAtEarliestInputTimestamp()));
-
- private final TimerInternals timerInternals;
- private final WindowingStrategy<?, W> windowingStrategy;
- private final StateTag<Object, WatermarkHoldState<W>> elementHoldTag;
-
- public WatermarkHold(TimerInternals timerInternals, WindowingStrategy<?, W> windowingStrategy) {
- this.timerInternals = timerInternals;
- this.windowingStrategy = windowingStrategy;
- this.elementHoldTag = watermarkHoldTagForOutputTimeFn(windowingStrategy.getOutputTimeFn());
- }
-
- /**
- * Add a hold to prevent the output watermark progressing beyond the (possibly adjusted) timestamp
- * of the element in {@code context}. We allow the actual hold time to be shifted later by
- * {@link OutputTimeFn#assignOutputTime}, but no further than the end of the window. The hold will
- * remain until cleared by {@link #extractAndRelease}. Return the timestamp at which the hold
- * was placed, or {@literal null} if no hold was placed.
- *
- * <p>In the following we'll write {@code E} to represent an element's timestamp after passing
- * through the window strategy's output time function, {@code IWM} for the local input watermark,
- * {@code OWM} for the local output watermark, and {@code GCWM} for the garbage collection
- * watermark (which is at {@code IWM - getAllowedLateness}). Time progresses from left to right,
- * and we write {@code [ ... ]} to denote a bounded window with implied lower bound.
- *
- * <p>Note that the GCWM will be the same as the IWM if {@code getAllowedLateness}
- * is {@code ZERO}.
- *
- * <p>Here are the cases we need to handle. They are conceptually considered in the
- * sequence written since if getAllowedLateness is ZERO the GCWM is the same as the IWM.
- * <ol>
- * <li>(Normal)
- * <pre>
- * |
- * [ | E ]
- * |
- * IWM
- * </pre>
- * This is, hopefully, the common and happy case. The element is locally on-time and can
- * definitely make it to an {@code ON_TIME} pane which we can still set an end-of-window timer
- * for. We place an element hold at E, which may contribute to the {@code ON_TIME} pane's
- * timestamp (depending on the output time function). Thus the OWM will not proceed past E
- * until the next pane fires.
- *
- * <li>(Discard - no target window)
- * <pre>
- * | |
- * [ E ] | |
- * | |
- * GCWM <-getAllowedLateness-> IWM
- * </pre>
- * The element is very locally late. The window has been garbage collected, thus there
- * is no target pane E could be assigned to. We discard E.
- *
- * <li>(Unobservably late)
- * <pre>
- * | |
- * [ | E | ]
- * | |
- * OWM IWM
- * </pre>
- * The element is locally late, however we can still treat this case as for 'Normal' above
- * since the IWM has not yet passed the end of the window and the element is ahead of the
- * OWM. In effect, we get to 'launder' the locally late element and consider it as locally
- * on-time because no downstream computation can observe the difference.
- *
- * <li>(Maybe late 1)
- * <pre>
- * | |
- * [ | E ] |
- * | |
- * OWM IWM
- * </pre>
- * The end-of-window timer may have already fired for this window, and thus an {@code ON_TIME}
- * pane may have already been emitted. However, if timer firings have been delayed then it
- * is possible the {@code ON_TIME} pane has not yet been emitted. We can't place an element
- * hold since we can't be sure if it will be cleared promptly. Thus this element *may* find
- * its way into an {@code ON_TIME} pane, but if so it will *not* contribute to that pane's
- * timestamp. We may however set a garbage collection hold if required.
- *
- * <li>(Maybe late 2)
- * <pre>
- * | |
- * [ E | | ]
- * | |
- * OWM IWM
- * </pre>
- * The end-of-window timer has not yet fired, so this element may still appear in an
- * {@code ON_TIME} pane. However the element is too late to contribute to the output
- * watermark hold, and thus won't contribute to the pane's timestamp. We can still place an
- * end-of-window hold.
- *
- * <li>(Maybe late 3)
- * <pre>
- * | |
- * [ E | ] |
- * | |
- * OWM IWM
- * </pre>
- * As for the (Maybe late 2) case, however we don't even know if the end-of-window timer
- * has already fired, or it is about to fire. We can place only the garbage collection hold,
- * if required.
- *
- * <li>(Definitely late)
- * <pre>
- * | |
- * [ E ] | |
- * | |
- * OWM IWM
- * </pre>
- * The element is definitely too late to make an {@code ON_TIME} pane. We are too late to
- * place an end-of-window hold. We can still place a garbage collection hold if required.
- *
- * </ol>
- */
- @Nullable
- public Instant addHolds(ReduceFn<?, ?, ?, W>.ProcessValueContext context) {
- Instant hold = addElementHold(context);
- if (hold == null) {
- hold = addEndOfWindowOrGarbageCollectionHolds(context);
- }
- return hold;
- }
-
- /**
- * Return {@code timestamp}, possibly shifted forward in time according to the window
- * strategy's output time function.
- */
- private Instant shift(Instant timestamp, W window) {
- Instant shifted = windowingStrategy.getOutputTimeFn().assignOutputTime(timestamp, window);
- if (shifted.isBefore(timestamp)) {
- throw new IllegalStateException(
- String.format("OutputTimeFn moved element from %s to earlier time %s for window %s",
- timestamp, shifted, window));
- }
- if (!timestamp.isAfter(window.maxTimestamp()) && shifted.isAfter(window.maxTimestamp())) {
- throw new IllegalStateException(
- String.format("OutputTimeFn moved element from %s to %s which is beyond end of window %s",
- timestamp, shifted, window));
- }
-
- return shifted;
- }
-
- /**
- * Add an element hold if possible. Return instant at which hold was added, or {@literal null}
- * if no hold was added.
- */
- @Nullable
- private Instant addElementHold(ReduceFn<?, ?, ?, W>.ProcessValueContext context) {
- // Give the window function a chance to move the hold timestamp forward to encourage progress.
- // (A later hold implies less impediment to the output watermark making progress, which in
- // turn encourages end-of-window triggers to fire earlier in following computations.)
- Instant elementHold = shift(context.timestamp(), context.window());
-
- Instant outputWM = timerInternals.currentOutputWatermarkTime();
- Instant inputWM = timerInternals.currentInputWatermarkTime();
-
- // Only add the hold if we can be sure:
- // - the backend will be able to respect it
- // (ie the hold is at or ahead of the output watermark), AND
- // - a timer will be set to clear it by the end of window
- // (ie the end of window is at or ahead of the input watermark).
- String which;
- boolean tooLate;
- // TODO: These case labels could be tightened.
- // See the case analysis in addHolds above for the motivation.
- if (outputWM != null && elementHold.isBefore(outputWM)) {
- which = "too late to effect output watermark";
- tooLate = true;
- } else if (inputWM != null && context.window().maxTimestamp().isBefore(inputWM)) {
- which = "too late for end-of-window timer";
- tooLate = true;
- } else {
- which = "on time";
- tooLate = false;
- context.state().access(elementHoldTag).add(elementHold);
- }
- WindowTracing.trace(
- "WatermarkHold.addHolds: element hold at {} is {} for "
- + "key:{}; window:{}; inputWatermark:{}; outputWatermark:{}",
- elementHold, which, context.key(), context.window(), inputWM,
- outputWM);
-
- return tooLate ? null : elementHold;
- }
-
- /**
- * Add an end-of-window hold or, if too late for that, a garbage collection hold (if required).
- * Return the {@link Instant} at which hold was added, or {@literal null} if no hold was added.
- *
- * <p>The end-of-window hold guarantees that an empty {@code ON_TIME} pane can be given
- * a timestamp which will not be considered beyond allowed lateness by any downstream computation.
- */
- @Nullable
- private Instant addEndOfWindowOrGarbageCollectionHolds(ReduceFn<?, ?, ?, W>.Context context) {
- Instant hold = addEndOfWindowHold(context);
- if (hold == null) {
- hold = addGarbageCollectionHold(context);
- }
- return hold;
- }
-
- /**
- * Add an end-of-window hold. Return the {@link Instant} at which hold was added,
- * or {@literal null} if no hold was added.
- *
- * <p>The end-of-window hold guarantees that any empty {@code ON_TIME} pane can be given
- * a timestamp which will not be considered beyond allowed lateness by any downstream computation.
- */
- @Nullable
- private Instant addEndOfWindowHold(ReduceFn<?, ?, ?, W>.Context context) {
- // Only add an end-of-window hold if we can be sure a timer will be set to clear it
- // by the end of window (ie the end of window is at or ahead of the input watermark).
- Instant outputWM = timerInternals.currentOutputWatermarkTime();
- Instant inputWM = timerInternals.currentInputWatermarkTime();
- String which;
- boolean tooLate;
- Instant eowHold = context.window().maxTimestamp();
- if (inputWM != null && eowHold.isBefore(inputWM)) {
- which = "too late for end-of-window timer";
- tooLate = true;
- } else {
- which = "on time";
- tooLate = false;
- Preconditions.checkState(outputWM == null || !eowHold.isBefore(outputWM),
- "End-of-window hold %s cannot be before output watermark %s", eowHold, outputWM);
- context.state().access(EXTRA_HOLD_TAG).add(eowHold);
- }
- WindowTracing.trace(
- "WatermarkHold.addEndOfWindowHold: end-of-window hold at {} is {} for "
- + "key:{}; window:{}; inputWatermark:{}; outputWatermark:{}",
- eowHold, which, context.key(), context.window(), inputWM,
- outputWM);
-
- return tooLate ? null : eowHold;
- }
-
- /**
- * Add a garbage collection hold, if required. Return the {@link Instant} at which hold was added,
- * or {@literal null} if no hold was added.
- *
- * <p>The garbage collection hold gurantees that any empty final pane can be given
- * a timestamp which will not be considered beyond allowed lateness by any downstream
- * computation. If we are sure no empty final panes can be emitted then there's no need
- * for an additional hold.
- */
- @Nullable
- private Instant addGarbageCollectionHold(ReduceFn<?, ?, ?, W>.Context context) {
- // Only add a garbage collection hold if we may need to emit an empty pane
- // at garbage collection time, and garbage collection time is strictly after the
- // end of window. (All non-empty panes will have holds at their output
- // time derived from their incoming elements and no additional hold is required.)
- if (context.windowingStrategy().getClosingBehavior() == ClosingBehavior.FIRE_ALWAYS
- && windowingStrategy.getAllowedLateness().isLongerThan(Duration.ZERO)) {
- Instant gcHold = context.window().maxTimestamp().plus(windowingStrategy.getAllowedLateness());
- Instant outputWM = timerInternals.currentOutputWatermarkTime();
- Instant inputWM = timerInternals.currentInputWatermarkTime();
- WindowTracing.trace(
- "WatermarkHold.addGarbageCollectionHold: garbage collection at {} hold for "
- + "key:{}; window:{}; inputWatermark:{}; outputWatermark:{}",
- gcHold, context.key(), context.window(), inputWM, outputWM);
- Preconditions.checkState(inputWM == null || !gcHold.isBefore(inputWM),
- "Garbage collection hold %s cannot be before input watermark %s", gcHold, inputWM);
- context.state().access(EXTRA_HOLD_TAG).add(gcHold);
- return gcHold;
- } else {
- return null;
- }
- }
-
- /**
- * Prefetch watermark holds in preparation for merging.
- */
- public void prefetchOnMerge(MergingStateAccessor<?, W> state) {
- StateMerging.prefetchWatermarks(state, elementHoldTag);
- }
-
- /**
- * Updates the watermark hold when windows merge if it is possible the merged value does
- * not equal all of the existing holds. For example, if the new window implies a later
- * watermark hold, then earlier holds may be released.
- */
- public void onMerge(ReduceFn<?, ?, ?, W>.OnMergeContext context) {
- WindowTracing.debug("onMerge: for key:{}; window:{}; inputWatermark:{}; outputWatermark:{}",
- context.key(), context.window(), timerInternals.currentInputWatermarkTime(),
- timerInternals.currentOutputWatermarkTime());
- StateMerging.mergeWatermarks(context.state(), elementHoldTag, context.window());
- // If we had a cheap way to determine if we have an element hold then we could
- // avoid adding an unnecessary end-of-window or garbage collection hold.
- // Simply reading the above merged watermark would impose an additional read for the
- // common case that the active window has just one undelying state address window and
- // the hold depends on the min of the elemest timestamps.
- StateMerging.clear(context.state(), EXTRA_HOLD_TAG);
- addEndOfWindowOrGarbageCollectionHolds(context);
- }
-
- /**
- * Return (a future for) the earliest hold for {@code context}. Clear all the holds after
- * reading, but add/restore an end-of-window or garbage collection hold if required.
- *
- * <p>The returned timestamp is the output timestamp according to the {@link OutputTimeFn}
- * from the windowing strategy of this {@link WatermarkHold}, combined across all the non-late
- * elements in the current pane. If there is no such value the timestamp is the end
- * of the window.
- */
- public ReadableState<Instant> extractAndRelease(
- final ReduceFn<?, ?, ?, W>.Context context, final boolean isFinished) {
- WindowTracing.debug(
- "extractAndRelease: for key:{}; window:{}; inputWatermark:{}; outputWatermark:{}",
- context.key(), context.window(), timerInternals.currentInputWatermarkTime(),
- timerInternals.currentOutputWatermarkTime());
- final WatermarkHoldState<W> elementHoldState = context.state().access(elementHoldTag);
- final WatermarkHoldState<BoundedWindow> extraHoldState = context.state().access(EXTRA_HOLD_TAG);
- return new ReadableState<Instant>() {
- @Override
- public ReadableState<Instant> readLater() {
- elementHoldState.readLater();
- extraHoldState.readLater();
- return this;
- }
-
- @Override
- public Instant read() {
- // Read both the element and extra holds.
- Instant elementHold = elementHoldState.read();
- Instant extraHold = extraHoldState.read();
- Instant hold;
- // Find the minimum, accounting for null.
- if (elementHold == null) {
- hold = extraHold;
- } else if (extraHold == null) {
- hold = elementHold;
- } else if (elementHold.isBefore(extraHold)) {
- hold = elementHold;
- } else {
- hold = extraHold;
- }
- if (hold == null || hold.isAfter(context.window().maxTimestamp())) {
- // If no hold (eg because all elements came in behind the output watermark), or
- // the hold was for garbage collection, take the end of window as the result.
- WindowTracing.debug(
- "WatermarkHold.extractAndRelease.read: clipping from {} to end of window "
- + "for key:{}; window:{}",
- hold, context.key(), context.window());
- hold = context.window().maxTimestamp();
- }
- WindowTracing.debug("WatermarkHold.extractAndRelease.read: clearing for key:{}; window:{}",
- context.key(), context.window());
-
- // Clear the underlying state to allow the output watermark to progress.
- elementHoldState.clear();
- extraHoldState.clear();
-
- if (!isFinished) {
- // Only need to leave behind an end-of-window or garbage collection hold
- // if future elements will be processed.
- addEndOfWindowOrGarbageCollectionHolds(context);
- }
-
- return hold;
- }
- };
- }
-
- /**
- * Clear any remaining holds.
- */
- public void clearHolds(ReduceFn<?, ?, ?, W>.Context context) {
- WindowTracing.debug(
- "WatermarkHold.clearHolds: For key:{}; window:{}; inputWatermark:{}; outputWatermark:{}",
- context.key(), context.window(), timerInternals.currentInputWatermarkTime(),
- timerInternals.currentOutputWatermarkTime());
- context.state().access(elementHoldTag).clear();
- context.state().access(EXTRA_HOLD_TAG).clear();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Weighted.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Weighted.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Weighted.java
deleted file mode 100644
index c31ad7f..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/Weighted.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-/**
- * Interface representing an object that has a weight, in unspecified units.
- */
-public interface Weighted {
- /**
- * Returns the weight of the object.
- */
- long getWeight();
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WeightedValue.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WeightedValue.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WeightedValue.java
deleted file mode 100644
index 4a6e840..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WeightedValue.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-/**
- * A {@code T} with an accompanying weight. Units are unspecified.
- *
- * @param <T> the underlying type of object
- */
-public final class WeightedValue<T> implements Weighted {
-
- private final T value;
- private final long weight;
-
- private WeightedValue(T value, long weight) {
- this.value = value;
- this.weight = weight;
- }
-
- public static <T> WeightedValue<T> of(T value, long weight) {
- return new WeightedValue<>(value, weight);
- }
-
- public long getWeight() {
- return weight;
- }
-
- public T getValue() {
- return value;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowTracing.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowTracing.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowTracing.java
deleted file mode 100644
index 6ae2f42..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowTracing.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Logging for window operations. Generally only feasible to enable on hand-picked pipelines.
- */
-public final class WindowTracing {
- private static final Logger LOG = LoggerFactory.getLogger(WindowTracing.class);
-
- public static void debug(String format, Object... args) {
- LOG.debug(format, args);
- }
-
- @SuppressWarnings("unused")
- public static void trace(String format, Object... args) {
- LOG.trace(format, args);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowedValue.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowedValue.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowedValue.java
deleted file mode 100644
index 1e944e2..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowedValue.java
+++ /dev/null
@@ -1,720 +0,0 @@
-/*******************************************************************************
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import static com.google.cloud.dataflow.sdk.util.Structs.addBoolean;
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import com.google.cloud.dataflow.sdk.coders.Coder;
-import com.google.cloud.dataflow.sdk.coders.CoderException;
-import com.google.cloud.dataflow.sdk.coders.CollectionCoder;
-import com.google.cloud.dataflow.sdk.coders.InstantCoder;
-import com.google.cloud.dataflow.sdk.coders.StandardCoder;
-import com.google.cloud.dataflow.sdk.transforms.windowing.BoundedWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.PaneInfo;
-import com.google.cloud.dataflow.sdk.transforms.windowing.PaneInfo.PaneInfoCoder;
-import com.google.cloud.dataflow.sdk.util.common.ElementByteSizeObserver;
-import com.google.common.base.MoreObjects;
-import com.google.common.base.Preconditions;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import org.joda.time.Instant;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * An immutable triple of value, timestamp, and windows.
- *
- * @param <T> the type of the value
- */
-public abstract class WindowedValue<T> {
-
- protected final T value;
- protected final PaneInfo pane;
-
- /**
- * Returns a {@code WindowedValue} with the given value, timestamp,
- * and windows.
- */
- public static <T> WindowedValue<T> of(
- T value,
- Instant timestamp,
- Collection<? extends BoundedWindow> windows,
- PaneInfo pane) {
- Preconditions.checkNotNull(pane);
-
- if (windows.size() == 0 && BoundedWindow.TIMESTAMP_MIN_VALUE.equals(timestamp)) {
- return valueInEmptyWindows(value, pane);
- } else if (windows.size() == 1) {
- return of(value, timestamp, windows.iterator().next(), pane);
- } else {
- return new TimestampedValueInMultipleWindows<>(value, timestamp, windows, pane);
- }
- }
-
- /**
- * Returns a {@code WindowedValue} with the given value, timestamp, and window.
- */
- public static <T> WindowedValue<T> of(
- T value,
- Instant timestamp,
- BoundedWindow window,
- PaneInfo pane) {
- Preconditions.checkNotNull(pane);
-
- boolean isGlobal = GlobalWindow.INSTANCE.equals(window);
- if (isGlobal && BoundedWindow.TIMESTAMP_MIN_VALUE.equals(timestamp)) {
- return valueInGlobalWindow(value, pane);
- } else if (isGlobal) {
- return new TimestampedValueInGlobalWindow<>(value, timestamp, pane);
- } else {
- return new TimestampedValueInSingleWindow<>(value, timestamp, window, pane);
- }
- }
-
- /**
- * Returns a {@code WindowedValue} with the given value in the {@link GlobalWindow} using the
- * default timestamp and pane.
- */
- public static <T> WindowedValue<T> valueInGlobalWindow(T value) {
- return new ValueInGlobalWindow<>(value, PaneInfo.NO_FIRING);
- }
-
- /**
- * Returns a {@code WindowedValue} with the given value in the {@link GlobalWindow} using the
- * default timestamp and the specified pane.
- */
- public static <T> WindowedValue<T> valueInGlobalWindow(T value, PaneInfo pane) {
- return new ValueInGlobalWindow<>(value, pane);
- }
-
- /**
- * Returns a {@code WindowedValue} with the given value and timestamp,
- * {@code GlobalWindow} and default pane.
- */
- public static <T> WindowedValue<T> timestampedValueInGlobalWindow(T value, Instant timestamp) {
- if (BoundedWindow.TIMESTAMP_MIN_VALUE.equals(timestamp)) {
- return valueInGlobalWindow(value);
- } else {
- return new TimestampedValueInGlobalWindow<>(value, timestamp, PaneInfo.NO_FIRING);
- }
- }
-
- /**
- * Returns a {@code WindowedValue} with the given value in no windows, and the default timestamp
- * and pane.
- */
- public static <T> WindowedValue<T> valueInEmptyWindows(T value) {
- return new ValueInEmptyWindows<T>(value, PaneInfo.NO_FIRING);
- }
-
- /**
- * Returns a {@code WindowedValue} with the given value in no windows, and the default timestamp
- * and the specified pane.
- */
- public static <T> WindowedValue<T> valueInEmptyWindows(T value, PaneInfo pane) {
- return new ValueInEmptyWindows<T>(value, pane);
- }
-
- private WindowedValue(T value, PaneInfo pane) {
- this.value = value;
- this.pane = checkNotNull(pane);
- }
-
- /**
- * Returns a new {@code WindowedValue} that is a copy of this one, but with a different value,
- * which may have a new type {@code NewT}.
- */
- public abstract <NewT> WindowedValue<NewT> withValue(NewT value);
-
- /**
- * Returns the value of this {@code WindowedValue}.
- */
- public T getValue() {
- return value;
- }
-
- /**
- * Returns the timestamp of this {@code WindowedValue}.
- */
- public abstract Instant getTimestamp();
-
- /**
- * Returns the windows of this {@code WindowedValue}.
- */
- public abstract Collection<? extends BoundedWindow> getWindows();
-
- /**
- * Returns the pane of this {@code WindowedValue} in its window.
- */
- public PaneInfo getPane() {
- return pane;
- }
-
- @Override
- public abstract boolean equals(Object o);
-
- @Override
- public abstract int hashCode();
-
- @Override
- public abstract String toString();
-
- private static final Collection<? extends BoundedWindow> GLOBAL_WINDOWS =
- Collections.singletonList(GlobalWindow.INSTANCE);
-
- /**
- * The abstract superclass of WindowedValue representations where
- * timestamp == MIN.
- */
- private abstract static class MinTimestampWindowedValue<T>
- extends WindowedValue<T> {
- public MinTimestampWindowedValue(T value, PaneInfo pane) {
- super(value, pane);
- }
-
- @Override
- public Instant getTimestamp() {
- return BoundedWindow.TIMESTAMP_MIN_VALUE;
- }
- }
-
- /**
- * The representation of a WindowedValue where timestamp == MIN and
- * windows == {GlobalWindow}.
- */
- private static class ValueInGlobalWindow<T>
- extends MinTimestampWindowedValue<T> {
- public ValueInGlobalWindow(T value, PaneInfo pane) {
- super(value, pane);
- }
-
- @Override
- public <NewT> WindowedValue<NewT> withValue(NewT value) {
- return new ValueInGlobalWindow<>(value, pane);
- }
-
- @Override
- public Collection<? extends BoundedWindow> getWindows() {
- return GLOBAL_WINDOWS;
- }
-
- @Override
- public boolean equals(Object o) {
- if (o instanceof ValueInGlobalWindow) {
- ValueInGlobalWindow<?> that = (ValueInGlobalWindow<?>) o;
- return Objects.equals(that.pane, this.pane)
- && Objects.equals(that.value, this.value);
- } else {
- return false;
- }
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(value, pane);
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(getClass())
- .add("value", value)
- .add("pane", pane)
- .toString();
- }
- }
-
- /**
- * The representation of a WindowedValue where timestamp == MIN and
- * windows == {}.
- */
- private static class ValueInEmptyWindows<T>
- extends MinTimestampWindowedValue<T> {
- public ValueInEmptyWindows(T value, PaneInfo pane) {
- super(value, pane);
- }
-
- @Override
- public <NewT> WindowedValue<NewT> withValue(NewT value) {
- return new ValueInEmptyWindows<>(value, pane);
- }
-
- @Override
- public Collection<? extends BoundedWindow> getWindows() {
- return Collections.emptyList();
- }
-
- @Override
- public boolean equals(Object o) {
- if (o instanceof ValueInEmptyWindows) {
- ValueInEmptyWindows<?> that = (ValueInEmptyWindows<?>) o;
- return Objects.equals(that.pane, this.pane)
- && Objects.equals(that.value, this.value);
- } else {
- return false;
- }
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(value, pane);
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(getClass())
- .add("value", value)
- .add("pane", pane)
- .toString();
- }
- }
-
- /**
- * The abstract superclass of WindowedValue representations where
- * timestamp is arbitrary.
- */
- private abstract static class TimestampedWindowedValue<T>
- extends WindowedValue<T> {
- protected final Instant timestamp;
-
- public TimestampedWindowedValue(T value,
- Instant timestamp,
- PaneInfo pane) {
- super(value, pane);
- this.timestamp = checkNotNull(timestamp);
- }
-
- @Override
- public Instant getTimestamp() {
- return timestamp;
- }
- }
-
- /**
- * The representation of a WindowedValue where timestamp {@code >}
- * MIN and windows == {GlobalWindow}.
- */
- private static class TimestampedValueInGlobalWindow<T>
- extends TimestampedWindowedValue<T> {
- public TimestampedValueInGlobalWindow(T value,
- Instant timestamp,
- PaneInfo pane) {
- super(value, timestamp, pane);
- }
-
- @Override
- public <NewT> WindowedValue<NewT> withValue(NewT value) {
- return new TimestampedValueInGlobalWindow<>(value, timestamp, pane);
- }
-
- @Override
- public Collection<? extends BoundedWindow> getWindows() {
- return GLOBAL_WINDOWS;
- }
-
- @Override
- public boolean equals(Object o) {
- if (o instanceof TimestampedValueInGlobalWindow) {
- TimestampedValueInGlobalWindow<?> that =
- (TimestampedValueInGlobalWindow<?>) o;
- return this.timestamp.isEqual(that.timestamp) // don't compare chronology objects
- && Objects.equals(that.pane, this.pane)
- && Objects.equals(that.value, this.value);
- } else {
- return false;
- }
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(value, pane, timestamp.getMillis());
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(getClass())
- .add("value", value)
- .add("timestamp", timestamp)
- .add("pane", pane)
- .toString();
- }
- }
-
- /**
- * The representation of a WindowedValue where timestamp is arbitrary and
- * windows == a single non-Global window.
- */
- private static class TimestampedValueInSingleWindow<T>
- extends TimestampedWindowedValue<T> {
- private final BoundedWindow window;
-
- public TimestampedValueInSingleWindow(T value,
- Instant timestamp,
- BoundedWindow window,
- PaneInfo pane) {
- super(value, timestamp, pane);
- this.window = checkNotNull(window);
- }
-
- @Override
- public <NewT> WindowedValue<NewT> withValue(NewT value) {
- return new TimestampedValueInSingleWindow<>(value, timestamp, window, pane);
- }
-
- @Override
- public Collection<? extends BoundedWindow> getWindows() {
- return Collections.singletonList(window);
- }
-
- @Override
- public boolean equals(Object o) {
- if (o instanceof TimestampedValueInSingleWindow) {
- TimestampedValueInSingleWindow<?> that =
- (TimestampedValueInSingleWindow<?>) o;
- return Objects.equals(that.value, this.value)
- && this.timestamp.isEqual(that.timestamp) // don't compare chronology objects
- && Objects.equals(that.pane, this.pane)
- && Objects.equals(that.window, this.window);
- } else {
- return false;
- }
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(value, timestamp.getMillis(), pane, window);
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(getClass())
- .add("value", value)
- .add("timestamp", timestamp)
- .add("window", window)
- .add("pane", pane)
- .toString();
- }
- }
-
- /**
- * The representation of a WindowedValue, excluding the special
- * cases captured above.
- */
- private static class TimestampedValueInMultipleWindows<T>
- extends TimestampedWindowedValue<T> {
- private Collection<? extends BoundedWindow> windows;
-
- public TimestampedValueInMultipleWindows(
- T value,
- Instant timestamp,
- Collection<? extends BoundedWindow> windows,
- PaneInfo pane) {
- super(value, timestamp, pane);
- this.windows = checkNotNull(windows);
- }
-
- @Override
- public <NewT> WindowedValue<NewT> withValue(NewT value) {
- return new TimestampedValueInMultipleWindows<>(value, timestamp, windows, pane);
- }
-
- @Override
- public Collection<? extends BoundedWindow> getWindows() {
- return windows;
- }
-
- @Override
- public boolean equals(Object o) {
- if (o instanceof TimestampedValueInMultipleWindows) {
- TimestampedValueInMultipleWindows<?> that =
- (TimestampedValueInMultipleWindows<?>) o;
- if (this.timestamp.isEqual(that.timestamp) // don't compare chronology objects
- && Objects.equals(that.value, this.value)
- && Objects.equals(that.pane, this.pane)) {
- ensureWindowsAreASet();
- that.ensureWindowsAreASet();
- return that.windows.equals(this.windows);
- }
- }
- return false;
- }
-
- @Override
- public int hashCode() {
- ensureWindowsAreASet();
- return Objects.hash(value, timestamp.getMillis(), pane, windows);
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(getClass())
- .add("value", value)
- .add("timestamp", timestamp)
- .add("windows", windows)
- .add("pane", pane)
- .toString();
- }
-
- private void ensureWindowsAreASet() {
- if (!(windows instanceof Set)) {
- windows = new LinkedHashSet<>(windows);
- }
- }
- }
-
-
- /////////////////////////////////////////////////////////////////////////////
-
- /**
- * Returns the {@code Coder} to use for a {@code WindowedValue<T>},
- * using the given valueCoder and windowCoder.
- */
- public static <T> FullWindowedValueCoder<T> getFullCoder(
- Coder<T> valueCoder,
- Coder<? extends BoundedWindow> windowCoder) {
- return FullWindowedValueCoder.of(valueCoder, windowCoder);
- }
-
- /**
- * Returns the {@code ValueOnlyCoder} from the given valueCoder.
- */
- public static <T> ValueOnlyWindowedValueCoder<T> getValueOnlyCoder(Coder<T> valueCoder) {
- return ValueOnlyWindowedValueCoder.of(valueCoder);
- }
-
- /**
- * Abstract class for {@code WindowedValue} coder.
- */
- public abstract static class WindowedValueCoder<T>
- extends StandardCoder<WindowedValue<T>> {
- final Coder<T> valueCoder;
-
- WindowedValueCoder(Coder<T> valueCoder) {
- this.valueCoder = checkNotNull(valueCoder);
- }
-
- /**
- * Returns the value coder.
- */
- public Coder<T> getValueCoder() {
- return valueCoder;
- }
-
- /**
- * Returns a new {@code WindowedValueCoder} that is a copy of this one,
- * but with a different value coder.
- */
- public abstract <NewT> WindowedValueCoder<NewT> withValueCoder(Coder<NewT> valueCoder);
- }
-
- /**
- * Coder for {@code WindowedValue}.
- */
- public static class FullWindowedValueCoder<T> extends WindowedValueCoder<T> {
- private final Coder<? extends BoundedWindow> windowCoder;
- // Precompute and cache the coder for a list of windows.
- private final Coder<Collection<? extends BoundedWindow>> windowsCoder;
-
- public static <T> FullWindowedValueCoder<T> of(
- Coder<T> valueCoder,
- Coder<? extends BoundedWindow> windowCoder) {
- return new FullWindowedValueCoder<>(valueCoder, windowCoder);
- }
-
- @JsonCreator
- public static FullWindowedValueCoder<?> of(
- @JsonProperty(PropertyNames.COMPONENT_ENCODINGS)
- List<Coder<?>> components) {
- checkArgument(components.size() == 2,
- "Expecting 2 components, got " + components.size());
- @SuppressWarnings("unchecked")
- Coder<? extends BoundedWindow> window = (Coder<? extends BoundedWindow>) components.get(1);
- return of(components.get(0), window);
- }
-
- FullWindowedValueCoder(Coder<T> valueCoder,
- Coder<? extends BoundedWindow> windowCoder) {
- super(valueCoder);
- this.windowCoder = checkNotNull(windowCoder);
- // It's not possible to statically type-check correct use of the
- // windowCoder (we have to ensure externally that we only get
- // windows of the class handled by windowCoder), so type
- // windowsCoder in a way that makes encode() and decode() work
- // right, and cast the window type away here.
- @SuppressWarnings({"unchecked", "rawtypes"})
- Coder<Collection<? extends BoundedWindow>> collectionCoder =
- (Coder) CollectionCoder.of(this.windowCoder);
- this.windowsCoder = collectionCoder;
- }
-
- public Coder<? extends BoundedWindow> getWindowCoder() {
- return windowCoder;
- }
-
- public Coder<Collection<? extends BoundedWindow>> getWindowsCoder() {
- return windowsCoder;
- }
-
- @Override
- public <NewT> WindowedValueCoder<NewT> withValueCoder(Coder<NewT> valueCoder) {
- return new FullWindowedValueCoder<>(valueCoder, windowCoder);
- }
-
- @Override
- public void encode(WindowedValue<T> windowedElem,
- OutputStream outStream,
- Context context)
- throws CoderException, IOException {
- Context nestedContext = context.nested();
- valueCoder.encode(windowedElem.getValue(), outStream, nestedContext);
- InstantCoder.of().encode(
- windowedElem.getTimestamp(), outStream, nestedContext);
- windowsCoder.encode(windowedElem.getWindows(), outStream, nestedContext);
- PaneInfoCoder.INSTANCE.encode(windowedElem.getPane(), outStream, context);
- }
-
- @Override
- public WindowedValue<T> decode(InputStream inStream, Context context)
- throws CoderException, IOException {
- Context nestedContext = context.nested();
- T value = valueCoder.decode(inStream, nestedContext);
- Instant timestamp = InstantCoder.of().decode(inStream, nestedContext);
- Collection<? extends BoundedWindow> windows =
- windowsCoder.decode(inStream, nestedContext);
- PaneInfo pane = PaneInfoCoder.INSTANCE.decode(inStream, nestedContext);
- return WindowedValue.of(value, timestamp, windows, pane);
- }
-
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- verifyDeterministic(
- "FullWindowedValueCoder requires a deterministic valueCoder",
- valueCoder);
- verifyDeterministic(
- "FullWindowedValueCoder requires a deterministic windowCoder",
- windowCoder);
- }
-
- @Override
- public void registerByteSizeObserver(WindowedValue<T> value,
- ElementByteSizeObserver observer,
- Context context) throws Exception {
- valueCoder.registerByteSizeObserver(value.getValue(), observer, context);
- InstantCoder.of().registerByteSizeObserver(value.getTimestamp(), observer, context);
- windowsCoder.registerByteSizeObserver(value.getWindows(), observer, context);
- }
-
- @Override
- public CloudObject asCloudObject() {
- CloudObject result = super.asCloudObject();
- addBoolean(result, PropertyNames.IS_WRAPPER, true);
- return result;
- }
-
- @Override
- public List<? extends Coder<?>> getCoderArguments() {
- return null;
- }
-
- @Override
- public List<? extends Coder<?>> getComponents() {
- return Arrays.<Coder<?>>asList(valueCoder, windowCoder);
- }
- }
-
- /**
- * Coder for {@code WindowedValue}.
- *
- * <p>A {@code ValueOnlyWindowedValueCoder} only encodes and decodes the value. It drops
- * timestamp and windows for encoding, and uses defaults timestamp, and windows for decoding.
- */
- public static class ValueOnlyWindowedValueCoder<T> extends WindowedValueCoder<T> {
- public static <T> ValueOnlyWindowedValueCoder<T> of(
- Coder<T> valueCoder) {
- return new ValueOnlyWindowedValueCoder<>(valueCoder);
- }
-
- @JsonCreator
- public static ValueOnlyWindowedValueCoder<?> of(
- @JsonProperty(PropertyNames.COMPONENT_ENCODINGS)
- List<Coder<?>> components) {
- checkArgument(components.size() == 1, "Expecting 1 component, got " + components.size());
- return of(components.get(0));
- }
-
- ValueOnlyWindowedValueCoder(Coder<T> valueCoder) {
- super(valueCoder);
- }
-
- @Override
- public <NewT> WindowedValueCoder<NewT> withValueCoder(Coder<NewT> valueCoder) {
- return new ValueOnlyWindowedValueCoder<>(valueCoder);
- }
-
- @Override
- public void encode(WindowedValue<T> windowedElem, OutputStream outStream, Context context)
- throws CoderException, IOException {
- valueCoder.encode(windowedElem.getValue(), outStream, context);
- }
-
- @Override
- public WindowedValue<T> decode(InputStream inStream, Context context)
- throws CoderException, IOException {
- T value = valueCoder.decode(inStream, context);
- return WindowedValue.valueInGlobalWindow(value);
- }
-
- @Override
- public void verifyDeterministic() throws NonDeterministicException {
- verifyDeterministic(
- "ValueOnlyWindowedValueCoder requires a deterministic valueCoder",
- valueCoder);
- }
-
- @Override
- public void registerByteSizeObserver(
- WindowedValue<T> value, ElementByteSizeObserver observer, Context context)
- throws Exception {
- valueCoder.registerByteSizeObserver(value.getValue(), observer, context);
- }
-
- @Override
- public CloudObject asCloudObject() {
- CloudObject result = super.asCloudObject();
- addBoolean(result, PropertyNames.IS_WRAPPER, true);
- return result;
- }
-
- @Override
- public List<? extends Coder<?>> getCoderArguments() {
- return Arrays.<Coder<?>>asList(valueCoder);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingInternals.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingInternals.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingInternals.java
deleted file mode 100644
index 12fcd53..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingInternals.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import com.google.cloud.dataflow.sdk.coders.Coder;
-import com.google.cloud.dataflow.sdk.transforms.windowing.BoundedWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.PaneInfo;
-import com.google.cloud.dataflow.sdk.util.state.StateInternals;
-import com.google.cloud.dataflow.sdk.values.PCollectionView;
-import com.google.cloud.dataflow.sdk.values.TupleTag;
-
-import org.joda.time.Instant;
-
-import java.io.IOException;
-import java.util.Collection;
-
-/**
- * Interface that may be required by some (internal) {@code DoFn}s to implement windowing. It should
- * not be necessary for general user code to interact with this at all.
- *
- * <p>This interface should be provided by runner implementors to support windowing on their runner.
- *
- * @param <InputT> input type
- * @param <OutputT> output type
- */
-public interface WindowingInternals<InputT, OutputT> {
-
- /**
- * Unsupported state internals. The key type is unknown. It is up to the user to use the
- * correct type of key.
- */
- StateInternals<?> stateInternals();
-
- /**
- * Output the value at the specified timestamp in the listed windows.
- */
- void outputWindowedValue(OutputT output, Instant timestamp,
- Collection<? extends BoundedWindow> windows, PaneInfo pane);
-
- /**
- * Return the timer manager provided by the underlying system, or null if Timers need
- * to be emulated.
- */
- TimerInternals timerInternals();
-
- /**
- * Access the windows the element is being processed in without "exploding" it.
- */
- Collection<? extends BoundedWindow> windows();
-
- /**
- * Access the pane of the current window(s).
- */
- PaneInfo pane();
-
- /**
- * Write the given {@link PCollectionView} data to a location accessible by other workers.
- */
- <T> void writePCollectionViewData(
- TupleTag<?> tag,
- Iterable<WindowedValue<T>> data,
- Coder<T> elemCoder) throws IOException;
-
- /**
- * Return the value of the side input for the window of a main input element.
- */
- <T> T sideInput(PCollectionView<T> view, BoundedWindow mainInputWindow);
-}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/7bef2b7e/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java
----------------------------------------------------------------------
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java
deleted file mode 100644
index c167b8c..0000000
--- a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java
+++ /dev/null
@@ -1,268 +0,0 @@
-/*
- * Copyright (C) 2015 Google Inc.
- *
- * Licensed 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 com.google.cloud.dataflow.sdk.util;
-
-import com.google.cloud.dataflow.sdk.annotations.Experimental;
-import com.google.cloud.dataflow.sdk.transforms.windowing.BoundedWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.DefaultTrigger;
-import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
-import com.google.cloud.dataflow.sdk.transforms.windowing.OutputTimeFn;
-import com.google.cloud.dataflow.sdk.transforms.windowing.Trigger;
-import com.google.cloud.dataflow.sdk.transforms.windowing.Window.ClosingBehavior;
-import com.google.cloud.dataflow.sdk.transforms.windowing.WindowFn;
-import com.google.common.base.MoreObjects;
-
-import org.joda.time.Duration;
-
-import java.io.Serializable;
-import java.util.Objects;
-
-/**
- * A {@code WindowingStrategy} describes the windowing behavior for a specific collection of values.
- * It has both a {@link WindowFn} describing how elements are assigned to windows and a
- * {@link Trigger} that controls when output is produced for each window.
- *
- * @param <T> type of elements being windowed
- * @param <W> {@link BoundedWindow} subclass used to represent the
- * windows used by this {@code WindowingStrategy}
- */
-public class WindowingStrategy<T, W extends BoundedWindow> implements Serializable {
-
- /**
- * The accumulation modes that can be used with windowing.
- */
- public enum AccumulationMode {
- DISCARDING_FIRED_PANES,
- ACCUMULATING_FIRED_PANES;
- }
-
- private static final Duration DEFAULT_ALLOWED_LATENESS = Duration.ZERO;
- private static final WindowingStrategy<Object, GlobalWindow> DEFAULT = of(new GlobalWindows());
-
- private final WindowFn<T, W> windowFn;
- private final OutputTimeFn<? super W> outputTimeFn;
- private final ExecutableTrigger<W> trigger;
- private final AccumulationMode mode;
- private final Duration allowedLateness;
- private final ClosingBehavior closingBehavior;
- private final boolean triggerSpecified;
- private final boolean modeSpecified;
- private final boolean allowedLatenessSpecified;
- private final boolean outputTimeFnSpecified;
-
- private WindowingStrategy(
- WindowFn<T, W> windowFn,
- ExecutableTrigger<W> trigger, boolean triggerSpecified,
- AccumulationMode mode, boolean modeSpecified,
- Duration allowedLateness, boolean allowedLatenessSpecified,
- OutputTimeFn<? super W> outputTimeFn, boolean outputTimeFnSpecified,
- ClosingBehavior closingBehavior) {
- this.windowFn = windowFn;
- this.trigger = trigger;
- this.triggerSpecified = triggerSpecified;
- this.mode = mode;
- this.modeSpecified = modeSpecified;
- this.allowedLateness = allowedLateness;
- this.allowedLatenessSpecified = allowedLatenessSpecified;
- this.closingBehavior = closingBehavior;
- this.outputTimeFn = outputTimeFn;
- this.outputTimeFnSpecified = outputTimeFnSpecified;
- }
-
- /**
- * Return a fully specified, default windowing strategy.
- */
- public static WindowingStrategy<Object, GlobalWindow> globalDefault() {
- return DEFAULT;
- }
-
- public static <T, W extends BoundedWindow> WindowingStrategy<T, W> of(
- WindowFn<T, W> windowFn) {
- return new WindowingStrategy<>(windowFn,
- ExecutableTrigger.create(DefaultTrigger.<W>of()), false,
- AccumulationMode.DISCARDING_FIRED_PANES, false,
- DEFAULT_ALLOWED_LATENESS, false,
- windowFn.getOutputTimeFn(), false,
- ClosingBehavior.FIRE_IF_NON_EMPTY);
- }
-
- public WindowFn<T, W> getWindowFn() {
- return windowFn;
- }
-
- public ExecutableTrigger<W> getTrigger() {
- return trigger;
- }
-
- public boolean isTriggerSpecified() {
- return triggerSpecified;
- }
-
- public Duration getAllowedLateness() {
- return allowedLateness;
- }
-
- public boolean isAllowedLatenessSpecified() {
- return allowedLatenessSpecified;
- }
-
- public AccumulationMode getMode() {
- return mode;
- }
-
- public boolean isModeSpecified() {
- return modeSpecified;
- }
-
- public ClosingBehavior getClosingBehavior() {
- return closingBehavior;
- }
-
- public OutputTimeFn<? super W> getOutputTimeFn() {
- return outputTimeFn;
- }
-
- public boolean isOutputTimeFnSpecified() {
- return outputTimeFnSpecified;
- }
-
- /**
- * Returns a {@link WindowingStrategy} identical to {@code this} but with the trigger set to
- * {@code wildcardTrigger}.
- */
- public WindowingStrategy<T, W> withTrigger(Trigger<?> wildcardTrigger) {
- @SuppressWarnings("unchecked")
- Trigger<W> typedTrigger = (Trigger<W>) wildcardTrigger;
- return new WindowingStrategy<T, W>(
- windowFn,
- ExecutableTrigger.create(typedTrigger), true,
- mode, modeSpecified,
- allowedLateness, allowedLatenessSpecified,
- outputTimeFn, outputTimeFnSpecified,
- closingBehavior);
- }
-
- /**
- * Returns a {@link WindowingStrategy} identical to {@code this} but with the accumulation mode
- * set to {@code mode}.
- */
- public WindowingStrategy<T, W> withMode(AccumulationMode mode) {
- return new WindowingStrategy<T, W>(
- windowFn,
- trigger, triggerSpecified,
- mode, true,
- allowedLateness, allowedLatenessSpecified,
- outputTimeFn, outputTimeFnSpecified,
- closingBehavior);
- }
-
- /**
- * Returns a {@link WindowingStrategy} identical to {@code this} but with the window function
- * set to {@code wildcardWindowFn}.
- */
- public WindowingStrategy<T, W> withWindowFn(WindowFn<?, ?> wildcardWindowFn) {
- @SuppressWarnings("unchecked")
- WindowFn<T, W> typedWindowFn = (WindowFn<T, W>) wildcardWindowFn;
-
- // The onus of type correctness falls on the callee.
- @SuppressWarnings("unchecked")
- OutputTimeFn<? super W> newOutputTimeFn = (OutputTimeFn<? super W>)
- (outputTimeFnSpecified ? outputTimeFn : typedWindowFn.getOutputTimeFn());
-
- return new WindowingStrategy<T, W>(
- typedWindowFn,
- trigger, triggerSpecified,
- mode, modeSpecified,
- allowedLateness, allowedLatenessSpecified,
- newOutputTimeFn, outputTimeFnSpecified,
- closingBehavior);
- }
-
- /**
- * Returns a {@link WindowingStrategy} identical to {@code this} but with the allowed lateness
- * set to {@code allowedLateness}.
- */
- public WindowingStrategy<T, W> withAllowedLateness(Duration allowedLateness) {
- return new WindowingStrategy<T, W>(
- windowFn,
- trigger, triggerSpecified,
- mode, modeSpecified,
- allowedLateness, true,
- outputTimeFn, outputTimeFnSpecified,
- closingBehavior);
- }
-
- public WindowingStrategy<T, W> withClosingBehavior(ClosingBehavior closingBehavior) {
- return new WindowingStrategy<T, W>(
- windowFn,
- trigger, triggerSpecified,
- mode, modeSpecified,
- allowedLateness, allowedLatenessSpecified,
- outputTimeFn, outputTimeFnSpecified,
- closingBehavior);
- }
-
- @Experimental(Experimental.Kind.OUTPUT_TIME)
- public WindowingStrategy<T, W> withOutputTimeFn(OutputTimeFn<?> outputTimeFn) {
-
- @SuppressWarnings("unchecked")
- OutputTimeFn<? super W> typedOutputTimeFn = (OutputTimeFn<? super W>) outputTimeFn;
-
- return new WindowingStrategy<T, W>(
- windowFn,
- trigger, triggerSpecified,
- mode, modeSpecified,
- allowedLateness, allowedLatenessSpecified,
- typedOutputTimeFn, true,
- closingBehavior);
- }
-
- @Override
- public String toString() {
- return MoreObjects.toStringHelper(this)
- .add("windowFn", windowFn)
- .add("allowedLateness", allowedLateness)
- .add("trigger", trigger)
- .add("accumulationMode", mode)
- .add("outputTimeFn", outputTimeFn)
- .toString();
- }
-
- @Override
- public boolean equals(Object object) {
- if (!(object instanceof WindowingStrategy)) {
- return false;
- }
- WindowingStrategy<?, ?> other = (WindowingStrategy<?, ?>) object;
- return
- isTriggerSpecified() == other.isTriggerSpecified()
- && isAllowedLatenessSpecified() == other.isAllowedLatenessSpecified()
- && isModeSpecified() == other.isModeSpecified()
- && getMode().equals(other.getMode())
- && getAllowedLateness().equals(other.getAllowedLateness())
- && getClosingBehavior().equals(other.getClosingBehavior())
- && getTrigger().equals(other.getTrigger())
- && getWindowFn().equals(other.getWindowFn());
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(triggerSpecified, allowedLatenessSpecified, modeSpecified,
- windowFn, trigger, mode, allowedLateness, closingBehavior);
- }
-}