...

This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Under Discussion Accpted

Discussion thread:https://lists.apache.org/thread/ogo7ntmj8srdcko2h86vvd9djjsjfvcj 

Vote thread: https://lists.apache.org/thread/q4kn2g6tmc837ph2zvff40pgpmgzok3d

JIRA:

...

This meant that the returned result could be of two potential data types: plain V  or ValueAndTimestamp<V> . This was a source of inconsistency. For instance, querying a kv-store  with KeyQuery  would return a V  type, but querying a ts-kv-store  would yield a ValueAndTimestamp<V> . This behavior is unintuitive and potentially confusing for developers.

To address this inconsistencyensure consistency, we propose suggest that KeyQuery   should be restricted to querying kv-stores  only, ensuring that it always returns a plain V  type, making the behavior of the aforementioned code more predictable. Similarly, RangeQuery  should be dedicated to querying kv-stores , consistently returning only the plain V always return the plain V type, enhancing the predictability of the mentioned code. Likewise, RangeQuery should uniformly return the plain V KeyValueIterator.

For those requiring timestamped values from a ts-kv-store , we recommend introducing a new query type: TimestampedKeyQuery . This new query will specifically target ts-kv-stores  and will return ValueAndTimestamp<V> . Furthermore, to complement this, TimestampedRangeQuery  should be introduced to query ranges in ts-kv-stores , ensuring that the returned value always includes timestamps.

...

@Evolving
public final class TimestampedKeyQuery<K, V> implements Query<ValueAndTimestamp<V>> {

...

    /**
     * Creates a query that will retrieve the record 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
     */
    public static <K, V> TimestampedKeyQuery<K, V> withKey(final K key) 

    /**
     * Specifies that the cache should be skipped during query evaluation. This means, that the query will always
     * get forwarded to the underlying store.
     */
    public TimestampedKeyQuery<K, V> skipCache() 

    /**
     * The key that was specified for this query.
     */
    public K getKeykey() 

    /**
     * The flag whether to skip the cache or not during query evaluation.
     */
    public boolean isSkipCache() 
}

TimestampedRangeQuery : Tailored for ranges with timestamps, this query will return a KeyValueIterator<K, ValueAndTimestamp<V>>

According to KIP-968, this KIP introduces the public enum ResultOrder to determine whether keys are sorted in ascending or descending  or unordered order. Order is based on the serialized byte[] of the keys, not the 'logical' key order. employs the withDescendingKeys() and withAscendingKeys() methods to specify that the keys should be ordered in descending or ascending or unordered sequence, and the resultOrder() method to retrieve the value of enum value in  ResultOrder. I've incorporated these variables and methods into the TimestampedRangeQuery  class and modified some method inputs. As a result, we can now use withDescendingKeys() to obtain results in reverse order and use withAscendingKeys to obtain the result in ascending order. 

@Evolving
public final class TimestampedRangeQuery<K, V> implements Query<KeyValueIterator<K, ValueAndTimestamp<V>>> {

    ...

    /**
     * Interactive range query using a lower and upper bound to filter the keys returned.
     * @param lower The key that specifies the lower bound of the range
     * @param upper The key that specifies the upper bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> TimestampedRangeQuery<K, V> withRange(final K lower, final K upper) 

     

	/**
     * Interactive range query using an upper bound to filter the keys returned Determines if the serialized byte[] of the keys in ascending or descending or unordered order.
     * If both <K,V> are null, RangQuery returns a full range scanOrder is based on the serialized byte[] of the keys, not the 'logical' key order.
     * @param upper The key that specifies @return return the order of return records base on the upperserialized boundbyte[] of the range
keys (can be unordered, or *in @paramascending, <K>or Thein keydescending typeorder).
     */
 @param <V> The valuepublic type
ResultOrder resultOrder() {
   */
    public static <K, V> TimestampedRangeQuery<K, V> withUpperBound(final K upper)return order;
    }

    /**
     * InteractiveSet rangethe query usingto areturn lowerthe boundserialized tobyte[] filterof the keys returnedin descending order.
     * @paramOrder loweris The key that specifiesbased on the lowerserialized boundbyte[] of the range
keys, not the 'logical' key *order.
 @param <K> The key type
* @return a new RangeQuery *instance @paramwith <V>descending The value typeflag set.
     */
    public static <K, V> TimestampedRangeQuery<K, V> withLowerBoundwithDescendingKeys(final) K lower)

{
     /**
   return  * Interactive scan query that returns all records in the store.new TimestampedRangeQuery<>(this.lower, this.upper, ResultOrder.DESCENDING);
    }

    /**
     * @paramSet <K>the Thequery keyto type
return the serialized byte[] of *the @paramkeys <V>in The value typeAscending order.
     */
 Order is based publicon staticthe <K, V> TimestampedRangeQuery<K, V> withNoBounds() 

    /**serialized byte[] of the keys, not the 'logical' key order.
     * @return Thea lowernew boundRangeQuery ofinstance thewith query,ascending ifflag specifiedset.
     */
    public Optional<K> getLowerBoundTimestampedRangeQuery<K, V> withAscendingKeys() 
{
    /**
    return * The upper bound of the query, if specified
     */
    public Optional<K> getUpperBound() new TimestampedRangeQuery<>(this.lower, this.upper, ResultOrder.ASCENDING);
    }
   
}

According to KIP-968, we introduce a public enum ResultOrder.

ResultOrder enum
It helps with specifying the order of the returned results by the query.

package org.apache.kafka.streams.query;
  
public enum ResultOrder {
    ANY,
    ASCENDING,
    DESCENDING
}

Compatibility, Deprecation, and Migration Plan

...