mjsax commented on code in PR #21650:
URL: https://github.com/apache/kafka/pull/21650#discussion_r2893856159
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java:
##########
@@ -83,4 +96,139 @@ public void put(final K key, final ValueTimestampHeaders<V>
value, final long wi
protected Bytes keyBytes(final K key, final Headers headers) {
return Bytes.wrap(serdes.rawKey(key, headers));
}
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R> QueryResult<R> query(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final long start = time.nanoseconds();
+ final QueryResult<R> result;
+
+ if (query instanceof WindowKeyQuery) {
+ result = runWindowKeyQuery((WindowKeyQuery<K,
ValueTimestampHeaders<V>>) query, positionBound, config);
+ } else if (query instanceof WindowRangeQuery) {
+ result = runWindowRangeQuery((WindowRangeQuery<K,
ValueTimestampHeaders<V>>) query, positionBound, config);
+ } else {
+ result = super.query(query, positionBound, config);
+ }
+
+ if (config.isCollectExecutionInfo()) {
+ result.addExecutionInfo(
+ "Handled in " + getClass() + " with conversion to
ValueAndTimestamp in "
+ + (time.nanoseconds() - start) + "ns");
+ }
+ return result;
+ }
+
+ /**
+ * Handles WindowKeyQuery by creating a MeteredWindowStoreIterator with
conversion from
+ * ValueTimestampHeaders to ValueAndTimestamp.
+ */
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runWindowKeyQuery(final WindowKeyQuery<K,
ValueTimestampHeaders<V>> query,
+ final PositionBound
positionBound,
+ final QueryConfig config) {
Review Comment:
nit: indention
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java:
##########
@@ -83,4 +96,139 @@ public void put(final K key, final ValueTimestampHeaders<V>
value, final long wi
protected Bytes keyBytes(final K key, final Headers headers) {
return Bytes.wrap(serdes.rawKey(key, headers));
}
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R> QueryResult<R> query(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final long start = time.nanoseconds();
+ final QueryResult<R> result;
+
+ if (query instanceof WindowKeyQuery) {
+ result = runWindowKeyQuery((WindowKeyQuery<K,
ValueTimestampHeaders<V>>) query, positionBound, config);
+ } else if (query instanceof WindowRangeQuery) {
+ result = runWindowRangeQuery((WindowRangeQuery<K,
ValueTimestampHeaders<V>>) query, positionBound, config);
+ } else {
+ result = super.query(query, positionBound, config);
Review Comment:
```suggestion
result = wrapped().query(query, positionBound, config);
```
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStoreWithHeaders.java:
##########
@@ -83,4 +96,139 @@ public void put(final K key, final ValueTimestampHeaders<V>
value, final long wi
protected Bytes keyBytes(final K key, final Headers headers) {
return Bytes.wrap(serdes.rawKey(key, headers));
}
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R> QueryResult<R> query(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final long start = time.nanoseconds();
+ final QueryResult<R> result;
+
+ if (query instanceof WindowKeyQuery) {
+ result = runWindowKeyQuery((WindowKeyQuery<K,
ValueTimestampHeaders<V>>) query, positionBound, config);
+ } else if (query instanceof WindowRangeQuery) {
+ result = runWindowRangeQuery((WindowRangeQuery<K,
ValueTimestampHeaders<V>>) query, positionBound, config);
+ } else {
+ result = super.query(query, positionBound, config);
+ }
+
+ if (config.isCollectExecutionInfo()) {
+ result.addExecutionInfo(
+ "Handled in " + getClass() + " with conversion to
ValueAndTimestamp in "
+ + (time.nanoseconds() - start) + "ns");
+ }
+ return result;
+ }
+
+ /**
+ * Handles WindowKeyQuery by creating a MeteredWindowStoreIterator with
conversion from
+ * ValueTimestampHeaders to ValueAndTimestamp.
+ */
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runWindowKeyQuery(final WindowKeyQuery<K,
ValueTimestampHeaders<V>> query,
+ final PositionBound
positionBound,
+ final QueryConfig config) {
+ final QueryResult<R> queryResult;
+ if (query.getTimeFrom().isPresent() && query.getTimeTo().isPresent()) {
+ final WindowKeyQuery<Bytes, byte[]> rawKeyQuery =
+ WindowKeyQuery.withKeyAndWindowStartRange(
+ keyBytes(query.getKey(), new RecordHeaders()),
+ query.getTimeFrom().get(),
+ query.getTimeTo().get()
+ );
+ final QueryResult<WindowStoreIterator<byte[]>> rawResult =
wrapped().query(
+ rawKeyQuery,
+ positionBound,
+ config
+ );
+ if (rawResult.isSuccess()) {
+ final Function<byte[], ValueAndTimestamp<V>> valueFrom = bytes
-> {
+ final ValueTimestampHeaders<V> vth =
serdes.valueFrom(bytes, new RecordHeaders());
+ return vth == null ? null :
ValueAndTimestamp.make(vth.value(), vth.timestamp());
+ };
+
+ final MeteredWindowStoreIterator<ValueAndTimestamp<V>>
typedResult =
+ new MeteredWindowStoreIterator<>(
+ rawResult.getResult(),
+ fetchSensor,
+ iteratorDurationSensor,
+ streamsMetrics,
+ valueFrom,
+ time,
+ numOpenIterators,
+ openIterators
+ );
+ final
QueryResult<MeteredWindowStoreIterator<ValueAndTimestamp<V>>> typedQueryResult =
+
InternalQueryResultUtil.copyAndSubstituteDeserializedResult(rawResult,
typedResult);
+ queryResult = (QueryResult<R>) typedQueryResult;
+ } else {
+ queryResult = (QueryResult<R>) rawResult;
+ }
+ } else {
+ queryResult = QueryResult.forFailure(
+ FailureReason.UNKNOWN_QUERY_TYPE,
+ "This store (" + getClass() + ") doesn't know how to"
+ + " execute the given query (" + query + ")
because"
+ + " WindowStores only supports
WindowKeyQuery.withKeyAndWindowStartRange."
Review Comment:
This error message seems incorrect. Cf `MeteredWindowStore#runKeyQuery(...)`
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java:
##########
@@ -211,7 +211,7 @@ public <R> QueryResult<R> query(final Query<R> query,
final PositionBound positionBound,
final QueryConfig config) {
- throw new UnsupportedOperationException("Queries (IQv2) are not
supported for timestamped window stores with headers yet.");
+ return store.query(query, positionBound, config);
Review Comment:
If this is an adaptor store, so we need to make some type translation here?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedWindowStoreWithHeadersBuilder.java:
##########
@@ -218,7 +218,7 @@ public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardAll() {
public <R> QueryResult<R> query(final Query<R> query,
final PositionBound positionBound,
final QueryConfig config) {
- throw new UnsupportedOperationException("Queries (IQv2) are not
supported for timestamped window stores with headers yet.");
+ return wrapped().query(query, positionBound, config);
Review Comment:
I think we can just remove the whole override and re-use the impl from the
`Wrapper` store.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
