mjsax commented on code in PR #14596: URL: https://github.com/apache/kafka/pull/14596#discussion_r1373965049
########## streams/src/main/java/org/apache/kafka/streams/query/VersionedKeyQuery.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.kafka.streams.query; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * Interactive query for retrieving a single record from a versioned state store based on its key and timestamp. Review Comment: We should add `@param <K>` and `@parm <V>` to describe the generic types ########## streams/src/main/java/org/apache/kafka/streams/query/VersionedKeyQuery.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.kafka.streams.query; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * Interactive query for retrieving a single record from a versioned state store based on its key and timestamp. + */ +@Evolving +public final class VersionedKeyQuery<K, V> implements Query<VersionedRecord<V>> { + + private final K key; + private final Optional<Instant> asOfTimestamp; + + private VersionedKeyQuery(final K key, final Optional<Instant> asOfTimestamp) { + this.key = Objects.requireNonNull(key); + this.asOfTimestamp = asOfTimestamp; + } + + /** + * Creates a query that will retrieve the record from a versioned state store identified by {@code key} if it exists + * (or {@code null} otherwise). + * @param key The key to retrieve + * @param <K> The type of the key + * @param <V> The type of the value that will be retrieved + * @throws NullPointerException if @param key is null + */ + public static <K, V> VersionedKeyQuery<K, V> withKey(final K key) { + return new VersionedKeyQuery<>(key, Optional.empty()); + } + + /** + * Specifies the timestamp for the key query. The key query returns the record version for the specified timestamp. Review Comment: `record's` ? (not sure) ########## streams/src/main/java/org/apache/kafka/streams/query/VersionedKeyQuery.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.kafka.streams.query; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * Interactive query for retrieving a single record from a versioned state store based on its key and timestamp. + */ +@Evolving +public final class VersionedKeyQuery<K, V> implements Query<VersionedRecord<V>> { + + private final K key; + private final Optional<Instant> asOfTimestamp; + + private VersionedKeyQuery(final K key, final Optional<Instant> asOfTimestamp) { + this.key = Objects.requireNonNull(key); + this.asOfTimestamp = asOfTimestamp; + } + + /** + * Creates a query that will retrieve the record from a versioned state store identified by {@code key} if it exists + * (or {@code null} otherwise). Review Comment: Should we explain that by default we query for the "latest" version of the key, and refer to `asOf` as further explanation? ########## streams/src/main/java/org/apache/kafka/streams/query/VersionedKeyQuery.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.kafka.streams.query; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * Interactive query for retrieving a single record from a versioned state store based on its key and timestamp. + */ +@Evolving +public final class VersionedKeyQuery<K, V> implements Query<VersionedRecord<V>> { + + private final K key; + private final Optional<Instant> asOfTimestamp; + + private VersionedKeyQuery(final K key, final Optional<Instant> asOfTimestamp) { + this.key = Objects.requireNonNull(key); + this.asOfTimestamp = asOfTimestamp; + } + + /** + * Creates a query that will retrieve the record from a versioned state store identified by {@code key} if it exists + * (or {@code null} otherwise). + * @param key The key to retrieve + * @param <K> The type of the key + * @param <V> The type of the value that will be retrieved + * @throws NullPointerException if @param key is null + */ + public static <K, V> VersionedKeyQuery<K, V> withKey(final K key) { + return new VersionedKeyQuery<>(key, Optional.empty()); + } + + /** + * Specifies the timestamp for the key query. The key query returns the record version for the specified timestamp. + * (To be more precise: The key query returns the record with the greatest timestamp <= asOfTimestamp) + * if @param asOfTimestamp is null, it will be considered as Optional.empty() + * @param asOfTimestamp The as of timestamp for timestamp + */ + public VersionedKeyQuery<K, V> asOf(final Instant asOfTimestamp) { + if (asOfTimestamp == null) { + return new VersionedKeyQuery<>(key, Optional.empty()); + } + return new VersionedKeyQuery<>(key, Optional.of(asOfTimestamp)); + } + + /** + * The key that was specified for this query. Review Comment: We should add `@return` (even if somewhat redundant) for completeness (same for `asOfTimestamp` below). ########## streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedStoreQueryUtils.java: ########## @@ -0,0 +1,189 @@ +/* + * 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.kafka.streams.state.internals; + +import static org.apache.kafka.common.utils.Utils.mkEntry; +import static org.apache.kafka.common.utils.Utils.mkMap; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.time.temporal.ChronoField; +import java.util.Map; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.StateStoreContext; +import org.apache.kafka.streams.query.FailureReason; +import org.apache.kafka.streams.query.Position; +import org.apache.kafka.streams.query.PositionBound; +import org.apache.kafka.streams.query.Query; +import org.apache.kafka.streams.query.QueryConfig; +import org.apache.kafka.streams.query.QueryResult; +import org.apache.kafka.streams.query.VersionedKeyQuery; +import org.apache.kafka.streams.state.StateSerdes; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.VersionedRecord; + +public final class VersionedStoreQueryUtils { + + /** + * a utility interface to facilitate stores' query dispatch logic, + * allowing them to generically store query execution logic as the values + * in a map. + */ + @FunctionalInterface + public interface QueryHandler { + QueryResult<?> apply( + final Query<?> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store + ); + } + + @SuppressWarnings("rawtypes") + private static final Map<Class, QueryHandler> QUERY_HANDLER_MAP = + mkMap( + mkEntry( + VersionedKeyQuery.class, + VersionedStoreQueryUtils::runVersionedKeyQuery + ) + ); + + // make this class uninstantiable + + private VersionedStoreQueryUtils() { + } + @SuppressWarnings("unchecked") + public static <R> QueryResult<R> handleBasicQueries( + final Query<R> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store, + final Position position, + final StateStoreContext context + ) { + + final long start = config.isCollectExecutionInfo() ? System.nanoTime() : -1L; + final QueryResult<R> result; + + final QueryHandler handler = QUERY_HANDLER_MAP.get(query.getClass()); + if (handler == null) { + result = QueryResult.forUnknownQueryType(query, store); + } else if (context == null || !isPermitted(position, positionBound, context.taskId().partition())) { + result = QueryResult.notUpToBound( + position, + positionBound, + context == null ? null : context.taskId().partition() + ); + } else { + result = (QueryResult<R>) handler.apply( + query, + positionBound, + config, + store + ); + } + if (config.isCollectExecutionInfo()) { + result.addExecutionInfo( + "Handled in " + store.getClass() + " in " + (System.nanoTime() - start) + "ns" + ); + } + result.setPosition(position); + return result; + } + + public static boolean isPermitted( + final Position position, + final PositionBound positionBound, + final int partition + ) { + final Position bound = positionBound.position(); + for (final String topic : bound.getTopics()) { + final Map<Integer, Long> partitionBounds = bound.getPartitionPositions(topic); + final Map<Integer, Long> seenPartitionPositions = position.getPartitionPositions(topic); + if (!partitionBounds.containsKey(partition)) { + // this topic isn't bounded for our partition, so just skip over it. + } else { + if (!seenPartitionPositions.containsKey(partition)) { + // we haven't seen a partition that we have a bound for + return false; + } else if (seenPartitionPositions.get(partition) < partitionBounds.get(partition)) { + // our current position is behind the bound + return false; + } + } + } + return true; + } + + @SuppressWarnings("unchecked") + private static <R> QueryResult<R> runVersionedKeyQuery(final Query<R> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store) { + if (store instanceof VersionedKeyValueStore) { + final VersionedKeyValueStore<Bytes, byte[]> versionedKeyValueStore = + (VersionedKeyValueStore<Bytes, byte[]>) store; + if (query instanceof VersionedKeyQuery) { + final VersionedKeyQuery<Bytes, byte[]> rawKeyQuery = + (VersionedKeyQuery<Bytes, byte[]>) query; + try { + final VersionedRecord<byte[]> bytes; + if (((VersionedKeyQuery<?, ?>) query).asOfTimestamp().isPresent()) { + bytes = versionedKeyValueStore.get(rawKeyQuery.key(), + ((VersionedKeyQuery<?, ?>) query).asOfTimestamp().get() + .getLong(ChronoField.INSTANT_SECONDS)); Review Comment: I think we need to use `toEpochMilli()` here to get the `Instant` converted to `long` correctly? ########## streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedStoreQueryUtils.java: ########## @@ -0,0 +1,189 @@ +/* + * 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.kafka.streams.state.internals; + +import static org.apache.kafka.common.utils.Utils.mkEntry; +import static org.apache.kafka.common.utils.Utils.mkMap; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.time.temporal.ChronoField; +import java.util.Map; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.StateStoreContext; +import org.apache.kafka.streams.query.FailureReason; +import org.apache.kafka.streams.query.Position; +import org.apache.kafka.streams.query.PositionBound; +import org.apache.kafka.streams.query.Query; +import org.apache.kafka.streams.query.QueryConfig; +import org.apache.kafka.streams.query.QueryResult; +import org.apache.kafka.streams.query.VersionedKeyQuery; +import org.apache.kafka.streams.state.StateSerdes; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.VersionedRecord; + +public final class VersionedStoreQueryUtils { + + /** + * a utility interface to facilitate stores' query dispatch logic, + * allowing them to generically store query execution logic as the values + * in a map. + */ + @FunctionalInterface + public interface QueryHandler { + QueryResult<?> apply( + final Query<?> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store + ); + } + + @SuppressWarnings("rawtypes") + private static final Map<Class, QueryHandler> QUERY_HANDLER_MAP = + mkMap( + mkEntry( + VersionedKeyQuery.class, + VersionedStoreQueryUtils::runVersionedKeyQuery + ) + ); + + // make this class uninstantiable + + private VersionedStoreQueryUtils() { + } + @SuppressWarnings("unchecked") + public static <R> QueryResult<R> handleBasicQueries( + final Query<R> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store, + final Position position, + final StateStoreContext context + ) { + + final long start = config.isCollectExecutionInfo() ? System.nanoTime() : -1L; + final QueryResult<R> result; + + final QueryHandler handler = QUERY_HANDLER_MAP.get(query.getClass()); + if (handler == null) { + result = QueryResult.forUnknownQueryType(query, store); + } else if (context == null || !isPermitted(position, positionBound, context.taskId().partition())) { + result = QueryResult.notUpToBound( + position, + positionBound, + context == null ? null : context.taskId().partition() + ); + } else { + result = (QueryResult<R>) handler.apply( + query, + positionBound, + config, + store + ); + } + if (config.isCollectExecutionInfo()) { + result.addExecutionInfo( + "Handled in " + store.getClass() + " in " + (System.nanoTime() - start) + "ns" + ); + } + result.setPosition(position); + return result; + } + + public static boolean isPermitted( + final Position position, + final PositionBound positionBound, + final int partition + ) { + final Position bound = positionBound.position(); + for (final String topic : bound.getTopics()) { + final Map<Integer, Long> partitionBounds = bound.getPartitionPositions(topic); + final Map<Integer, Long> seenPartitionPositions = position.getPartitionPositions(topic); + if (!partitionBounds.containsKey(partition)) { + // this topic isn't bounded for our partition, so just skip over it. + } else { + if (!seenPartitionPositions.containsKey(partition)) { + // we haven't seen a partition that we have a bound for + return false; + } else if (seenPartitionPositions.get(partition) < partitionBounds.get(partition)) { + // our current position is behind the bound + return false; + } + } + } + return true; + } + + @SuppressWarnings("unchecked") + private static <R> QueryResult<R> runVersionedKeyQuery(final Query<R> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store) { + if (store instanceof VersionedKeyValueStore) { + final VersionedKeyValueStore<Bytes, byte[]> versionedKeyValueStore = + (VersionedKeyValueStore<Bytes, byte[]>) store; + if (query instanceof VersionedKeyQuery) { + final VersionedKeyQuery<Bytes, byte[]> rawKeyQuery = + (VersionedKeyQuery<Bytes, byte[]>) query; + try { + final VersionedRecord<byte[]> bytes; + if (((VersionedKeyQuery<?, ?>) query).asOfTimestamp().isPresent()) { + bytes = versionedKeyValueStore.get(rawKeyQuery.key(), + ((VersionedKeyQuery<?, ?>) query).asOfTimestamp().get() + .getLong(ChronoField.INSTANT_SECONDS)); + } else { + bytes = versionedKeyValueStore.get(rawKeyQuery.key()); + } + return (QueryResult<R>) QueryResult.forResult(bytes); + } catch (final Exception e) { + final String message = parseStoreException(e, store, query); + return QueryResult.forFailure( + FailureReason.STORE_EXCEPTION, + message + ); + } + } else { + // Here is preserved for VersionedMultiTimestampedKeyQuery + return QueryResult.forUnknownQueryType(query, store); + } + } else { + return QueryResult.forUnknownQueryType(query, store); + } + } + + public static <V> VersionedRecord<V> deserializeVersionedRecord(final StateSerdes<?, V> serdes, final VersionedRecord<byte[]> rawVersionedRecord) { + + final Deserializer<V> valueDeserializer = serdes.valueDeserializer(); + final long timestamp = rawVersionedRecord.timestamp(); + final V value = valueDeserializer.deserialize(serdes.topic(), rawVersionedRecord.value()); + valueDeserializer.close(); Review Comment: `valueDesarializer()` does not create a new instance/object, but returns a shared object -- we should not close it. ########## streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java: ########## @@ -89,6 +102,18 @@ private class MeteredVersionedKeyValueStoreInternal private final Serde<V> plainValueSerde; private StateSerdes<K, V> plainValueSerdes; + private final Map<Class, QueryHandler> queryHandlers = + mkMap( + mkEntry( + RangeQuery.class, Review Comment: Ah I see -- but for this case, we also need to add `KeyQuery` and also keep the existing test. As mentioned elsewhere, we could add support for `KeyQuery` and `RangeQuery` in a follow up KIP, so for now we should keep the guard for both in place. ########## streams/src/main/java/org/apache/kafka/streams/query/VersionedKeyQuery.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.kafka.streams.query; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * Interactive query for retrieving a single record from a versioned state store based on its key and timestamp. + */ +@Evolving +public final class VersionedKeyQuery<K, V> implements Query<VersionedRecord<V>> { + + private final K key; + private final Optional<Instant> asOfTimestamp; + + private VersionedKeyQuery(final K key, final Optional<Instant> asOfTimestamp) { + this.key = Objects.requireNonNull(key); + this.asOfTimestamp = asOfTimestamp; + } + + /** + * Creates a query that will retrieve the record from a versioned state store identified by {@code key} if it exists + * (or {@code null} otherwise). + * @param key The key to retrieve + * @param <K> The type of the key + * @param <V> The type of the value that will be retrieved + * @throws NullPointerException if @param key is null + */ + public static <K, V> VersionedKeyQuery<K, V> withKey(final K key) { + return new VersionedKeyQuery<>(key, Optional.empty()); + } + + /** + * Specifies the timestamp for the key query. The key query returns the record version for the specified timestamp. + * (To be more precise: The key query returns the record with the greatest timestamp <= asOfTimestamp) Review Comment: Is `<=` valid or does it need HTML escaping? ########## streams/src/main/java/org/apache/kafka/streams/query/VersionedKeyQuery.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.kafka.streams.query; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * Interactive query for retrieving a single record from a versioned state store based on its key and timestamp. + */ +@Evolving +public final class VersionedKeyQuery<K, V> implements Query<VersionedRecord<V>> { + + private final K key; + private final Optional<Instant> asOfTimestamp; + + private VersionedKeyQuery(final K key, final Optional<Instant> asOfTimestamp) { + this.key = Objects.requireNonNull(key); + this.asOfTimestamp = asOfTimestamp; + } + + /** + * Creates a query that will retrieve the record from a versioned state store identified by {@code key} if it exists + * (or {@code null} otherwise). + * @param key The key to retrieve + * @param <K> The type of the key + * @param <V> The type of the value that will be retrieved + * @throws NullPointerException if @param key is null Review Comment: Is `@param key` valid JavaDoc? -- we usually use `{@code key}` for stuff like this IIRC. ########## streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedStoreQueryUtils.java: ########## @@ -0,0 +1,189 @@ +/* + * 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.kafka.streams.state.internals; + +import static org.apache.kafka.common.utils.Utils.mkEntry; +import static org.apache.kafka.common.utils.Utils.mkMap; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.time.temporal.ChronoField; +import java.util.Map; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.StateStoreContext; +import org.apache.kafka.streams.query.FailureReason; +import org.apache.kafka.streams.query.Position; +import org.apache.kafka.streams.query.PositionBound; +import org.apache.kafka.streams.query.Query; +import org.apache.kafka.streams.query.QueryConfig; +import org.apache.kafka.streams.query.QueryResult; +import org.apache.kafka.streams.query.VersionedKeyQuery; +import org.apache.kafka.streams.state.StateSerdes; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.VersionedRecord; + +public final class VersionedStoreQueryUtils { Review Comment: Why do we need a new class? Can't we add what we need to existing `StoreQueryUtils` (seems we introduce a lot of (unnecessary?) code duplication)? ########## streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java: ########## @@ -148,13 +199,33 @@ protected <R> QueryResult<R> runRangeQuery(final Query<R> query, throw new UnsupportedOperationException("Versioned stores do not support RangeQuery queries at this time."); } - @Override - protected <R> QueryResult<R> runKeyQuery(final Query<R> query, - final PositionBound positionBound, - final QueryConfig config) { - // throw exception for now to reserve the ability to implement this in the future - // without clashing with users' custom implementations in the meantime - throw new UnsupportedOperationException("Versioned stores do not support KeyQuery queries at this time."); + @SuppressWarnings("unchecked") + protected <R> QueryResult<R> runVersionedKeyQuery(final Query<R> query, + final PositionBound positionBound, + final QueryConfig config) { Review Comment: nit: fix indention ########## streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java: ########## @@ -148,13 +199,33 @@ protected <R> QueryResult<R> runRangeQuery(final Query<R> query, throw new UnsupportedOperationException("Versioned stores do not support RangeQuery queries at this time."); } - @Override - protected <R> QueryResult<R> runKeyQuery(final Query<R> query, - final PositionBound positionBound, - final QueryConfig config) { - // throw exception for now to reserve the ability to implement this in the future - // without clashing with users' custom implementations in the meantime - throw new UnsupportedOperationException("Versioned stores do not support KeyQuery queries at this time."); + @SuppressWarnings("unchecked") + protected <R> QueryResult<R> runVersionedKeyQuery(final Query<R> query, + final PositionBound positionBound, + final QueryConfig config) { + if (query instanceof VersionedKeyQuery) { + final QueryResult<R> result; + final VersionedKeyQuery<K, V> typedKeyQuery = (VersionedKeyQuery<K, V>) query; + VersionedKeyQuery<Bytes, byte[]> rawKeyQuery; + rawKeyQuery = VersionedKeyQuery.withKey(keyBytes(typedKeyQuery.key())); + if (typedKeyQuery.asOfTimestamp().isPresent()) { + rawKeyQuery = rawKeyQuery.asOf(typedKeyQuery.asOfTimestamp().get()); + } + final QueryResult<VersionedRecord<byte[]>> rawResult = + wrapped().query(rawKeyQuery, positionBound, config); + if (rawResult.isSuccess()) { + final VersionedRecord<V> versionedRecord = VersionedStoreQueryUtils.deserializeVersionedRecord(plainValueSerdes, rawResult.getResult()); + final QueryResult<VersionedRecord<V>> typedQueryResult = + InternalQueryResultUtil.copyAndSubstituteDeserializedResult(rawResult, versionedRecord); + result = (QueryResult<R>) typedQueryResult; + } else { + // the generic type doesn't matter, since failed queries have no result set. + result = (QueryResult<R>) rawResult; + } + return result; + } + // reserved for other IQv2 query types (KIP-968, KIP-969) Review Comment: Actually wondering if this is the right approach? If we add new query types, we would add new query handler and new methods, right? So we don't need the `if (query instanceof VersionedKeyQuery)` to begin with, because we would call `runVersionedKeyQuery` only if the type was `VerionedKeyQuery` to begin with (then query handler ensure this part)? ########## streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java: ########## @@ -148,13 +199,33 @@ protected <R> QueryResult<R> runRangeQuery(final Query<R> query, throw new UnsupportedOperationException("Versioned stores do not support RangeQuery queries at this time."); } - @Override - protected <R> QueryResult<R> runKeyQuery(final Query<R> query, - final PositionBound positionBound, - final QueryConfig config) { - // throw exception for now to reserve the ability to implement this in the future - // without clashing with users' custom implementations in the meantime - throw new UnsupportedOperationException("Versioned stores do not support KeyQuery queries at this time."); + @SuppressWarnings("unchecked") + protected <R> QueryResult<R> runVersionedKeyQuery(final Query<R> query, + final PositionBound positionBound, + final QueryConfig config) { + if (query instanceof VersionedKeyQuery) { + final QueryResult<R> result; + final VersionedKeyQuery<K, V> typedKeyQuery = (VersionedKeyQuery<K, V>) query; + VersionedKeyQuery<Bytes, byte[]> rawKeyQuery; + rawKeyQuery = VersionedKeyQuery.withKey(keyBytes(typedKeyQuery.key())); Review Comment: Merge this and the line above? ########## streams/src/main/java/org/apache/kafka/streams/query/VersionedKeyQuery.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.kafka.streams.query; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * Interactive query for retrieving a single record from a versioned state store based on its key and timestamp. + */ +@Evolving +public final class VersionedKeyQuery<K, V> implements Query<VersionedRecord<V>> { + + private final K key; + private final Optional<Instant> asOfTimestamp; + + private VersionedKeyQuery(final K key, final Optional<Instant> asOfTimestamp) { + this.key = Objects.requireNonNull(key); + this.asOfTimestamp = asOfTimestamp; + } + + /** + * Creates a query that will retrieve the record from a versioned state store identified by {@code key} if it exists + * (or {@code null} otherwise). + * @param key The key to retrieve + * @param <K> The type of the key + * @param <V> The type of the value that will be retrieved + * @throws NullPointerException if @param key is null + */ + public static <K, V> VersionedKeyQuery<K, V> withKey(final K key) { + return new VersionedKeyQuery<>(key, Optional.empty()); + } + + /** + * Specifies the timestamp for the key query. The key query returns the record version for the specified timestamp. + * (To be more precise: The key query returns the record with the greatest timestamp <= asOfTimestamp) + * if @param asOfTimestamp is null, it will be considered as Optional.empty() Review Comment: Did we discuss this in the KIP (cannot remember) -- by gut feeling would be that `null` should not be allowed (because not necessary, as "latest" is the default anyway)? ########## streams/src/main/java/org/apache/kafka/streams/query/VersionedKeyQuery.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.kafka.streams.query; + +import java.time.Instant; +import java.util.Objects; +import java.util.Optional; +import org.apache.kafka.common.annotation.InterfaceStability.Evolving; +import org.apache.kafka.streams.state.VersionedRecord; + +/** + * Interactive query for retrieving a single record from a versioned state store based on its key and timestamp. + */ +@Evolving +public final class VersionedKeyQuery<K, V> implements Query<VersionedRecord<V>> { + + private final K key; + private final Optional<Instant> asOfTimestamp; + + private VersionedKeyQuery(final K key, final Optional<Instant> asOfTimestamp) { + this.key = Objects.requireNonNull(key); + this.asOfTimestamp = asOfTimestamp; + } + + /** + * Creates a query that will retrieve the record from a versioned state store identified by {@code key} if it exists + * (or {@code null} otherwise). + * @param key The key to retrieve + * @param <K> The type of the key + * @param <V> The type of the value that will be retrieved + * @throws NullPointerException if @param key is null + */ + public static <K, V> VersionedKeyQuery<K, V> withKey(final K key) { + return new VersionedKeyQuery<>(key, Optional.empty()); + } + + /** + * Specifies the timestamp for the key query. The key query returns the record version for the specified timestamp. + * (To be more precise: The key query returns the record with the greatest timestamp <= asOfTimestamp) + * if @param asOfTimestamp is null, it will be considered as Optional.empty() + * @param asOfTimestamp The as of timestamp for timestamp Review Comment: nit `as-of` why `for timestamp` Or simpler: `The point in time for the query.` (Does not sound great either to be frank...) ########## streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedStoreQueryUtils.java: ########## @@ -0,0 +1,189 @@ +/* + * 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.kafka.streams.state.internals; + +import static org.apache.kafka.common.utils.Utils.mkEntry; +import static org.apache.kafka.common.utils.Utils.mkMap; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.time.temporal.ChronoField; +import java.util.Map; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.processor.StateStore; +import org.apache.kafka.streams.processor.StateStoreContext; +import org.apache.kafka.streams.query.FailureReason; +import org.apache.kafka.streams.query.Position; +import org.apache.kafka.streams.query.PositionBound; +import org.apache.kafka.streams.query.Query; +import org.apache.kafka.streams.query.QueryConfig; +import org.apache.kafka.streams.query.QueryResult; +import org.apache.kafka.streams.query.VersionedKeyQuery; +import org.apache.kafka.streams.state.StateSerdes; +import org.apache.kafka.streams.state.VersionedKeyValueStore; +import org.apache.kafka.streams.state.VersionedRecord; + +public final class VersionedStoreQueryUtils { + + /** + * a utility interface to facilitate stores' query dispatch logic, + * allowing them to generically store query execution logic as the values + * in a map. + */ + @FunctionalInterface + public interface QueryHandler { + QueryResult<?> apply( + final Query<?> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store + ); + } + + @SuppressWarnings("rawtypes") + private static final Map<Class, QueryHandler> QUERY_HANDLER_MAP = + mkMap( + mkEntry( + VersionedKeyQuery.class, + VersionedStoreQueryUtils::runVersionedKeyQuery + ) + ); + + // make this class uninstantiable + + private VersionedStoreQueryUtils() { + } + @SuppressWarnings("unchecked") + public static <R> QueryResult<R> handleBasicQueries( + final Query<R> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store, + final Position position, + final StateStoreContext context + ) { + + final long start = config.isCollectExecutionInfo() ? System.nanoTime() : -1L; + final QueryResult<R> result; + + final QueryHandler handler = QUERY_HANDLER_MAP.get(query.getClass()); + if (handler == null) { + result = QueryResult.forUnknownQueryType(query, store); + } else if (context == null || !isPermitted(position, positionBound, context.taskId().partition())) { + result = QueryResult.notUpToBound( + position, + positionBound, + context == null ? null : context.taskId().partition() + ); + } else { + result = (QueryResult<R>) handler.apply( + query, + positionBound, + config, + store + ); + } + if (config.isCollectExecutionInfo()) { + result.addExecutionInfo( + "Handled in " + store.getClass() + " in " + (System.nanoTime() - start) + "ns" + ); + } + result.setPosition(position); + return result; + } + + public static boolean isPermitted( + final Position position, + final PositionBound positionBound, + final int partition + ) { + final Position bound = positionBound.position(); + for (final String topic : bound.getTopics()) { + final Map<Integer, Long> partitionBounds = bound.getPartitionPositions(topic); + final Map<Integer, Long> seenPartitionPositions = position.getPartitionPositions(topic); + if (!partitionBounds.containsKey(partition)) { + // this topic isn't bounded for our partition, so just skip over it. + } else { + if (!seenPartitionPositions.containsKey(partition)) { + // we haven't seen a partition that we have a bound for + return false; + } else if (seenPartitionPositions.get(partition) < partitionBounds.get(partition)) { + // our current position is behind the bound + return false; + } + } + } + return true; + } + + @SuppressWarnings("unchecked") + private static <R> QueryResult<R> runVersionedKeyQuery(final Query<R> query, + final PositionBound positionBound, + final QueryConfig config, + final StateStore store) { Review Comment: nit: fix indention ########## streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java: ########## @@ -148,13 +199,33 @@ protected <R> QueryResult<R> runRangeQuery(final Query<R> query, throw new UnsupportedOperationException("Versioned stores do not support RangeQuery queries at this time."); } - @Override - protected <R> QueryResult<R> runKeyQuery(final Query<R> query, Review Comment: Related to above: we should keep this method ########## streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java: ########## @@ -148,13 +205,38 @@ protected <R> QueryResult<R> runRangeQuery(final Query<R> query, throw new UnsupportedOperationException("Versioned stores do not support RangeQuery queries at this time."); } + @SuppressWarnings("unchecked") @Override protected <R> QueryResult<R> runKeyQuery(final Query<R> query, - final PositionBound positionBound, - final QueryConfig config) { - // throw exception for now to reserve the ability to implement this in the future - // without clashing with users' custom implementations in the meantime - throw new UnsupportedOperationException("Versioned stores do not support KeyQuery queries at this time."); + final PositionBound positionBound, + final QueryConfig config) { + if (query instanceof VersionedKeyQuery) { + final QueryResult<R> result; + final VersionedKeyQuery<K, V> typedKeyQuery = (VersionedKeyQuery<K, V>) query; + VersionedKeyQuery<Bytes, byte[]> rawKeyQuery; + if (typedKeyQuery.asOfTimestamp().isPresent()) { + rawKeyQuery = VersionedKeyQuery.withKey(keyBytes(typedKeyQuery.key())); + rawKeyQuery = rawKeyQuery.asOf(typedKeyQuery.asOfTimestamp().get()); + } else { + rawKeyQuery = VersionedKeyQuery.withKey(keyBytes(typedKeyQuery.key())); + } + final QueryResult<VersionedRecord<byte[]>> rawResult = + wrapped().query(rawKeyQuery, positionBound, config); + if (rawResult.isSuccess()) { + final Function<byte[], ValueAndTimestamp<V>> deserializer = getDeserializeValue(plainValueSerdes); Review Comment: > MeteredVersionedKeyValueStoreInternal every other method uses ValueAndTimestamp as well. This part is for DSL compatibility only. All DSL operator only work with `ValueAndTimestamp` type right now, so when we plugin the new versioned state store, we need to translate between the types to make it work. > Btw, do you agree to implement a new class VersionedRecordDeserializer? As this point, I don't think we need such a class? The helper method you added does the job perfectly fine IMHO. But I am also not against it, if it make it easier (we would add it as internal class though only I assume?). ########## streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java: ########## @@ -148,13 +205,38 @@ protected <R> QueryResult<R> runRangeQuery(final Query<R> query, throw new UnsupportedOperationException("Versioned stores do not support RangeQuery queries at this time."); } + @SuppressWarnings("unchecked") @Override protected <R> QueryResult<R> runKeyQuery(final Query<R> query, Review Comment: > I think we do not have simple key queries with versioned state stores Not right now, but I think we could add support for `KeyQuery` and `RangeQuery` and target the "latest segment" only, ie `KeyQuery == VersionedKeyQuery(w/o asOfTimestamp)` in the future and should keep the door open for this. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
