[flink-statefun] 01/09: [FLINK-15954] Add PersistedTable SDK classes

[tz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git

commit 840b005bcf5dcd493519b3c3befb2233dd4f524c
Author: Igal Shilman <ig...@gmail.com>
AuthorDate: Fri Feb 7 15:24:11 2020 +0100

    [FLINK-15954] Add PersistedTable SDK classes
---
 .../flink/statefun/sdk/state/PersistedTable.java   | 148 +++++++++++++++++++++
 .../flink/statefun/sdk/state/TableAccessor.java    |  27 ++++
 2 files changed, 175 insertions(+)

diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedTable.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedTable.java
new file mode 100644
index 0000000..f54166f
--- /dev/null
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/PersistedTable.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.statefun.sdk.state;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.flink.statefun.sdk.StatefulFunction;
+import org.apache.flink.statefun.sdk.annotations.ForRuntime;
+import org.apache.flink.statefun.sdk.annotations.Persisted;
+
+/**
+ * A {@link PersistedTable} is a table (collection of keys and values) registered within {@link
+ * StatefulFunction}s and is persisted and maintained by the system for fault-tolerance.
+ *
+ * <p>Created persisted tables must be registered by using the {@link Persisted} annotation. Please
+ * see the class-level Javadoc of {@link StatefulFunction} for an example on how to do that.
+ *
+ * @see StatefulFunction
+ * @param <K> type of the key - Please note that the key have a meaningful {@link #hashCode()} and
+ *     {@link #equals(Object)} implemented.
+ * @param <V> type of the value.
+ */
+public final class PersistedTable<K, V> {
+  private final String name;
+  private final Class<K> keyType;
+  private final Class<V> valueType;
+  private TableAccessor<K, V> accessor;
+
+  private PersistedTable(
+      String name, Class<K> keyType, Class<V> valueType, TableAccessor<K, V> accessor) {
+    this.name = Objects.requireNonNull(name);
+    this.keyType = Objects.requireNonNull(keyType);
+    this.valueType = Objects.requireNonNull(valueType);
+    this.accessor = Objects.requireNonNull(accessor);
+  }
+
+  /**
+   * Creates a {@link PersistedTable} instance that may be used to access persisted state managed by
+   * the system. Access to the persisted table is identified by an unique name, type of the key, and
+   * type of the value. These may not change across multiple executions of the application.
+   *
+   * @param name the unique name of the persisted state.
+   * @param keyType the type of the state keys of this {@code PersistedTable}.
+   * @param valueType the type of the state values of this {@code PersistedTale}.
+   * @param <K> the type of the state keys.
+   * @param <V> the type of the state values.
+   * @return a {@code PersistedTable} instance.
+   */
+  public static <K, V> PersistedTable<K, V> of(String name, Class<K> keyType, Class<V> valueType) {
+    return new PersistedTable<>(name, keyType, valueType, new NonFaultTolerantAccessor<>());
+  }
+
+  /**
+   * Returns the unique name of the persisted table.
+   *
+   * @return unique name of the persisted table.
+   */
+  public String name() {
+    return name;
+  }
+
+  /**
+   * Returns the type of the persisted tables keys.
+   *
+   * @return the type of the persisted tables keys.
+   */
+  public Class<K> keyType() {
+    return keyType;
+  }
+
+  /**
+   * Returns the type of the persisted tables values.
+   *
+   * @return the type of the persisted tables values.
+   */
+  public Class<V> valueType() {
+    return valueType;
+  }
+
+  /**
+   * Returns a persisted table's value.
+   *
+   * @return the persisted table value associated with {@code key}.
+   */
+  public V get(K key) {
+    return accessor.get(key);
+  }
+
+  /**
+   * Updates the persisted table.
+   *
+   * @param key the to associate the value with.
+   * @param value the new value.
+   */
+  public void set(K key, V value) {
+    accessor.set(key, value);
+  }
+
+  /**
+   * Removes the value associated with {@code key}.
+   *
+   * @param key the key to remove.
+   */
+  public void remove(K key) {
+    accessor.remove(key);
+  }
+
+  @ForRuntime
+  void setAccessor(TableAccessor<K, V> newAccessor) {
+    Objects.requireNonNull(newAccessor);
+    this.accessor = newAccessor;
+  }
+
+  private static final class NonFaultTolerantAccessor<K, V> implements TableAccessor<K, V> {
+    private final Map<K, V> map = new HashMap<>();
+
+    @Override
+    public void set(K key, V value) {
+      map.put(key, value);
+    }
+
+    @Override
+    public V get(K key) {
+      return map.get(key);
+    }
+
+    @Override
+    public void remove(K key) {
+      map.remove(key);
+    }
+  }
+}
diff --git a/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/TableAccessor.java b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/TableAccessor.java
new file mode 100644
index 0000000..af68972
--- /dev/null
+++ b/statefun-sdk/src/main/java/org/apache/flink/statefun/sdk/state/TableAccessor.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.statefun.sdk.state;
+
+public interface TableAccessor<K, V> {
+
+  void set(K key, V value);
+
+  V get(K key);
+
+  void remove(K key);
+}