SAMZA-1423: Implement time series storage for joins and windows #303

vjagadish1989 · 2017-09-26T17:35:31Z

Notable changes:

New interface for storing and retrieving time series data.
New store and serde implementation for use in windows and joins

Pending:

Documentation, and minor clean-ups
Wire-up of stores from ExecutionPlanner
Usage of the store to implement various windows and joins

…store

vjagadish1989 · 2017-09-26T17:42:15Z

@prateekm @nickpan47 @jmakes @xinyuiscool - Please review :-)

jmakes

I'll try to take a deeper look later, but here were a couple things that stood out at first glance.

jmakes · 2017-09-26T18:47:54Z

samza-api/src/main/java/org/apache/samza/storage/kv/ClosableIterator.java

+ *
+ * @param <V> the type of value returned by this iterator
+ */
+public interface ClosableIterator<V> extends Iterator<V> {


Seems like it would be better to leverage the java Closeable interface so we can support AutoCloseable for try-with-resources
https://docs.oracle.com/javase/7/docs/api/java/lang/AutoCloseable.html

While the AutoClosable idiom is useful for file handles, it expects the caller to catch the checked Exception each time they call iterator.close(). Almost always, In Samza we 'd want to shutdown when there's an exception during iterator.close(), So, a runtime exception makes more sense.

Furthermore, iterator.close() on our existing KVStore API does not throw any checked Exception. I'd prefer to lean on the side of consistency, and refactor it later if there's enough consensus. Happy to discuss offline, and follow-up!

jmakes · 2017-09-26T18:53:03Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesValue.java

+ */
+public class TimeSeriesValue<V> {
+  private final V value;
+  private final Long timestamp;


Should this timestamp always match the timestamp in the key? If so, does it need to be in both the key and the value?

There are no such requirements.

On the write side, you'd call put(k, v, timestamp)

On the query side, you ask the store to give you all values for a key within a timestamp range. The return value is an iterator that returns the value along with the timestamp they were added to the store.

There's no need to store the timestamp in both the key and the value.

prateekm · 2017-09-26T17:50:01Z

samza-api/src/main/java/org/apache/samza/storage/kv/ClosableIterator.java

+/**
+ * An iterator that can be closed.
+ *
+ * <p> Implement close to free resources assigned to the iterator such as open file handles, persistent state etc.


Prefer putting the paragraph tag on the preceding empty line for readability, here and elsewhere.

good find , I will fix it uniformly everywhere! :)

prateekm · 2017-09-26T17:51:15Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKey.java

+  private final K key;
+  private final long timestamp;
+
+  // allows a maximum of 2 Billion entries per window per key. maybe, long?


int should be fine, can remove comment.

prateekm · 2017-09-26T17:51:57Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKey.java

+  @Override
+  public String toString() {
+    return "TimeSeriesKey{" +
+            "key=" + key +


Minor: Indentation. Maybe String.format?

Wil fix, thanks!

prateekm · 2017-09-26T17:55:17Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKeySerde.java

+    int seqNum = timeSeriesKey.getSeqNum();
+
+    byte[] serializedKey = keySerde.toBytes(key);
+    int keySize = serializedKey == null ? 0 : serializedKey.length;


Why do we allow null key bytes? Are there cases where timestamp + seq is sufficient?

Null bytes for key are needed to implement non-keyed tumbling and session windows. Basically, in that case, the start range of the timestamp will uniquely identify a window. The start range of the timestamp in combination with the sequence number will uniquely identify an entry in the store

prateekm · 2017-09-26T17:56:10Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStore.java

+ * @param <K> the type of key in the store
+ * @param <V> the type of value in the store
+ */
+


Minor: Delete newline.

prateekm · 2017-09-26T19:05:42Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStoreImpl.java

+    }
+
+    @Override
+    public TimeSeriesValue<V> next() {


Should this return the key as well? See question about range.

Since all access to the store are based on a key K and a timestamp, it's not necessary for the key to be returned again as a part of the iterator.

prateekm · 2017-09-26T19:06:29Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesValue.java

+ *
+ * @param <V> the type of the value
+ */
+public class TimeSeriesValue<V> {


Can we use the KV class from the serde PR when it's available?

Well, I ended up renaming the TimeSeriesValue<V> to TimestampedValue<V>.

In general, prefer new classes (with their own semantic meanings) over KV<K, V> or Pair<K,V>.

prateekm · 2017-09-26T19:34:20Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStoreImpl.java

+   * Since timestamps are at the granularity of milliseconds, multiple entries added in the same
+   * millisecond are distinguished by a monotonically increasing sequence number.
+   */
+  private int seqNum = 0;


There's a potential issue in this implementation with ordering of values for the same key in the same millisecond when the seqNum wraps.

I think we should be OK though: We get 2G seq nums before we wrap. Given that we have 8 byte timestamp + 4 byte seqNum overhead per key, this is at the very minimum (12 bytes * 2 G) = 24 GB for a store with empty keys and values. Even an (int, int) key value store would have 20 * 2 = 40 GB storage at a minimum (there's additional rocksDB ttl timestamp and write amplification overhead), and this is per TaskInstance.

Would prefer changing this to an AtomicInteger instead of a int + synchronization.

Just throwing it out there since we were also considering storing a version number with the key: with a 4 byte overhead per key, we can use 7 LSB from an AtomicLong seq num + a 1 byte version num to solve both issues.

Thank you for the suggestion.

After discussion we agreed the following:

The wrapping is a non-issue if we use an atomic long

We'll use a version for the MSB, and use the next 7 bytes for storing the long value in the sequence number

prateekm · 2017-09-26T19:35:36Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKey.java

+
+    TimeSeriesKey<?> that = (TimeSeriesKey<?>) o;
+
+    if (timestamp != that.timestamp) return false;


prateekm · 2017-09-26T19:38:24Z

samza-api/src/main/java/org/apache/samza/storage/kv/ClosableIterator.java

+ *
+ * @param <V> the type of value returned by this iterator
+ */
+public interface ClosableIterator<V> extends Iterator<V> {


Can we use KeyValueIterator instead?

We agreed that since all accesses to the store are by a particular key anyway, there's no need to return a KVIterator. The user of the store knows the key when methods are invoked.

prateekm · 2017-09-26T20:03:35Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStoreImpl.java

+  }
+
+  @Override
+  public void remove(K key, Long startTimestamp, Long endTimeStamp) {


Maybe add a remove(K key) too?

prateekm

LGTM, with one question about version mismatch handling. Thanks!

prateekm · 2017-09-28T00:52:27Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKey.java

+    return seqNum;
+  }
+
+


Delete extra newline.

prateekm · 2017-09-28T00:53:32Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKeySerde.java

+public class TimeSeriesKeySerde<K> implements Serde<TimeSeriesKey<K>> {
+
+  private static final long SEQUENCE_NUM_MASK = 0x00ffffffffffffffL;
+


Delete newline.

prateekm · 2017-09-28T00:54:27Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKeySerde.java

+    }
+
+    long timeStamp = buf.getLong();
+    long seqNum = buf.getLong();


This relies on version being 0. Safer to do the right thing here now.
Q: What should happen if the version doesn't match the field in the class? Should we throw here?

prateekm · 2017-09-28T00:56:42Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStore.java

+   * @param key the key to look up in the store
+   * @throws IllegalArgumentException when startTimeStamp &gt; endTimeStamp, or when either of them is negative
+   */
+  void remove(K key);


Maybe add the same version for get? Or can remove this one too, now that I know that we don't need to use this for join.

prateekm · 2017-09-28T00:57:22Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStoreImpl.java

+ *    +----------------------+--------8 bytes----+----1 bytes-+---------7 bytes--------+----value size----------
+ *    +----------------------------------STORE KEY-------------------------------------+---STORE VAL-----------+
+ *  </pre>
+ *  An 8 byte timestamp, a one byte version and a 7 byte sequence number are appended to the provided key and this


Minor: Indent lines 66-69 by -1.

Thanks for catching this!

prateekm · 2017-09-28T00:57:47Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStoreImpl.java

+ * <p>
+ * A {@link TimeSeriesStoreImpl} can be backed by persistent stores like rocksDB, in-memory stores, change-logged
+ * stores, cached stores (or any combination of these).
+ *


Minor: What I meant to say was that we don't need newline if the p tag is on its own line (here and everywhere else).

prateekm · 2017-09-28T00:59:13Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimestampedValue.java

+/**
+ * An immutable pair of a value, and its corresponding timestamp.
+ *
+ * <p> Iterators on {@link TimeSeriesStore}s always return {@link TimestampedValue}s


Can remove this sentence.

prateekm · 2017-09-28T01:07:24Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStoreImpl.java

+ *  An 8 byte timestamp, a one byte version and a 7 byte sequence number are appended to the provided key and this
+ *  combination is used as the key in the k-v store. The provided value is stored as is.
+ *
+ *  <p> This class is thread-safe and concurrent reads/writes are expected.


Does the KVStore API/impl provide read after write consistency across threads as well? I.e. do we have any guarantees about when a value written by T1 will be visible to T2?

The RocksDb docs state that the store is "thread-safe". We'll have to find out more on whether they mean:

Concurrent operations on the store will not corrupt state

Operations on the store "happen-before" other operations? That directly translates to visibility across threads.

[2] is a much stronger guarantee on [1]. I'd expect us to support [2]

prateekm · 2017-09-28T01:19:42Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStore.java

+   * Returns an iterator over values for the given key in the provided time-range - [{@code startTimestamp}, {@code endTimestamp})
+   *
+   * Values returned by the iterator are ordered by their timestamp. Values with the same timestamp are
+   * returned in their order of insertion. The iterator must be closed after use by calling {@link #close}.


IMHO the "Iterator must be closed" warning is worth putting on a newline and putting a <b> around must. Forgetting to do so will lead to memory leaks.

+1 for the extra emphasis

…store

xinyuiscool

LGTM. Have a few comments.

xinyuiscool · 2017-09-28T22:43:38Z

build.gradle

@@ -167,6 +167,7 @@ project(":samza-core_$scalaVersion") {
    compile "org.scala-lang:scala-library:$scalaLibVersion"
    compile "org.slf4j:slf4j-api:$slf4jVersion"
    testCompile project(":samza-api").sourceSets.test.output
+    testCompile project(":samza-kv-rocksdb_$scalaVersion")


Why core depends on kv-rocksDb? Seems this will cause circular dependency. I would prefer the test cases added to kv-rocksdb to avoid this.

That would mean that all tests for operators (window/ join) that instantiate stores would now be a part of samza-kv-rocksdb. Let's sync up offline since the scope of that seems to be larger than this PR.

Also, I believe circular dependencies in testCompile tasks are fine in gradle.

xinyuiscool · 2017-09-28T22:46:56Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKey.java

+    return timestamp;
+  }
+
+  public byte getVersion() {


seems a little bit weird that it's a member fn to return the static val.

Thanks, will fix it!

@xinyuiscool I accidentally committed this without addressing this feedback. However, I ended up doing a follow-up patch that addresses your comment - 92894e5

xinyuiscool · 2017-09-28T23:06:03Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStore.java

+ * @param <K> the type of key in the store
+ * @param <V> the type of value in the store
+ */
+public interface TimeSeriesStore<K, V> {


should V be something extends fromTimestampedValue?

Not really, <V> should be completely agnostic to types here for this to be a general purpose API. However, the return type from get should be an Iterator<TimeStampedValue<V>> (as it's being done)

xinyuiscool · 2017-09-28T23:19:59Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStoreImpl.java

+  }
+
+  @Override
+  public ClosableIterator<TimestampedValue<V>> get(K key, long startTimestamp, long endTimestamp) {


Why not close the iterator automatically when it reaches to the end? Now we have to remind user to close it after using.

Closing iterators automatically means that iterator.hasNext() is stateful - The invocation of hasNext immediately after the iterator has been closed will return an InvalidStateException. It also forces that hasNext can only return false once.

Alternately, If we choose to return false during hasNext(), we should have a way to distinguish these 2 cases:
false returned due to no elements to iterate upon
false returned due to iterator being closed.

For these reasons, and to maintain consistency with our already existing iterator APIs that are built on top of RocksDb, I'd prefer to avoid auto-closing iterators.

xinyuiscool · 2017-09-28T23:20:40Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesStoreImpl.java

+    }
+  }
+
+  private static class TimeSeriesStoreIterator<K, V> implements ClosableIterator<TimestampedValue<V>> {


as mentioned above, is it possible to automatically close it when it reaches the end?

prateekm · 2017-10-02T20:16:29Z

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKeySerde.java

+    long version = seqNum & ~SEQUENCE_NUM_MASK;
+
+    if (version != 0) {
+      throw new SamzaException("Version is not zero. Sequence number: " + seqNum);


This check should be version != TimeSeriesKey.VERSION.

Jagadish added 7 commits September 25, 2017 04:52

stuff

9420aa2

Merge branch 'master' of https://github.com/apache/samza into window-…

5f86534

…store

clean up APIs and add docs

625facc

clean up APIs and add docs

26517ce

revert unintended changes

d44a2df

revert unintended changes

7f3a91f

Minor cleanup, remove unnecessary spaces; Make checkstyle happy

37ff612

jmakes reviewed Sep 26, 2017

View reviewed changes

prateekm reviewed Sep 26, 2017

View reviewed changes

Jagadish added 7 commits September 27, 2017 16:51

window store pr

b713100

cleanups

1243686

Address Prateek's feedback

6e0b50b

Organize static methods together. Address Prateek's feedback

2adff46

Rename TimeSeriesvalue --> TimestampedValue

f75ac12

Minor cleanups; checkstyle fixes

87e8649

Minor cleanups; checkstyle fixes

665d2e0

prateekm approved these changes Sep 28, 2017

View reviewed changes

prateekm reviewed Sep 28, 2017

View reviewed changes

Merge branch 'master' of https://github.com/apache/samza into window-…

b6d5b7c

…store

xinyuiscool approved these changes Sep 28, 2017

View reviewed changes

Address review feedback

39dbfad

asfgit closed this in 56d564c Oct 2, 2017

prateekm reviewed Oct 2, 2017

View reviewed changes


		TimeSeriesKey<?> that = (TimeSeriesKey<?>) o;

		if (timestamp != that.timestamp) return false;

		public class TimeSeriesKeySerde<K> implements Serde<TimeSeriesKey<K>> {

		private static final long SEQUENCE_NUM_MASK = 0x00ffffffffffffffL;

SAMZA-1423: Implement time series storage for joins and windows #303

SAMZA-1423: Implement time series storage for joins and windows #303

Conversation

vjagadish1989 commented Sep 26, 2017

vjagadish1989 commented Sep 26, 2017

jmakes left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 Sep 28, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

prateekm Sep 26, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

prateekm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 Oct 2, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xinyuiscool left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 Oct 2, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 Oct 2, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjagadish1989 Sep 28, 2017 •

edited

prateekm Sep 26, 2017 •

edited

vjagadish1989 Oct 2, 2017 •

edited

vjagadish1989 Oct 2, 2017 •

edited

vjagadish1989 Oct 2, 2017 •

edited