package org.apache.kafka.streams.query;


/**
 * Interactive query for retrieving a set of records with keys within a specified key range and time
 * range.
 */

@Evolving
public final class MultiVersionedRangeQuery<K, V> implements
    Query<KeyValueIterator<K, VersionedRecord<V>>> {

  private final Optional<K> lower;
  private final Optional<K> upper;
  private final Optional<Instant> fromTimestampfromTime;

  private final Optional<Instant> asOfTimestamptoTime;

  private MultiVersionedRangeQuery(
 final boolean isKeyAscending;
  private final booleanOptional<K> isTimeAscending;
  private final boolean isOrderedByKey;

  private MultiVersionedRangeQuery(
      lower,
      final Optional<K> lowerupper,
      final Optional<K>Optional<Instant> upperfromTime,
      final Optional<Instant> fromTimestamp,toTime) {
    this.lower  final Optional<Instant> asOfTimestamp,= lower;
    this.upper  final boolean isOrderedByKey,= upper;
    this.fromTime = finalfromTime;
 boolean isKeyAscending,
  this.toTime = toTime;
  final}

  boolean isTimeAscending)/** {
    this.lower = lower;
    this.upper = upper;
    this.fromTimestamp = fromTimestamp;
    this.asOfTimestamp = asOfTimestamp;
    this.isOrderedByKey = isOrderedByKey;
    this.isKeyAscending = isKeyAscending;
    this.isTimeAscending = isTimeAscending;
  }

   /** 
   * Interactive range query using a lower and upper bound to filter the keys returned. * For each 
   * key the records valid within the specified time range are returned. * In case the time range is* Interactive range query using a lower and upper bound to filter the keys returned. * For each 
   * key the records valid within the specified time range are returned. * In case the time range is 
   * not specified just the latest record for each key is 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 
   * not@param specified<V> justThe thevalue latesttype record
 for each key is returned. */ 
   *public @paramstatic lower<K, TheV> keyMultiVersionedRangeQuery<K, thatV> specifieswithKeyRange(final theK lower, boundfinal of the range K upper);


   /**
   * @paramInteractive upperrange Thequery keyusing thata specifieslower thebound upperto bound offilter the range 
  keys returned. * @paramFor <K> Theeach key type the
   * @paramrecords <V>valid Thewithin valuethe typespecified 
time range are */ 
   public static <K, V> MultiVersionedRangeQuery<K, V> withKeyRange(final K lower, final K upper);


    /**returned. * In case the time range is not
   * specified just the latest record for each key is returned.
   * Interactive@param rangelower query using aThe key that specifies the lower bound toof filterthe therange
 keys returned. * @param <K>  For eachThe key thetype
   * records@param valid<V> within the specifiedThe timevalue rangetype
 are returned. */
 In casepublic thestatic time<K, rangeV> isMultiVersionedRangeQuery<K, not
   * specified just the latest record for each key is returned.
   * @param lower The key that specifies the lower bound of the rangeV> withLowerKeyBound(final K lower);

  /**
   * Interactive range query using a lower bound to filter the keys returned. * For each key the
   * @paramrecords <K>valid within the Thespecified keytime type
range are returned. * @paramIn <V>case the time Therange valueis typenot
   */
 specified publicjust staticthe <K,latest V>record MultiVersionedRangeQuery<K,for V>each withLowerKeyBound(finalkey K lower);

is returned.
   /**
 @param upper *The Interactivekey rangethat queryspecifies using athe lower bound toof filterthe therange
 keys returned. * @param <K> For  eachThe key thetype
   * records@param valid<V> within the specifiedThe timevalue rangetype
 are returned. */
 In casepublic thestatic time<K, rangeV> is not
   * specified just the latest record for each key is returned.
   * @param upper The key that specifies the lower bound of the rangeMultiVersionedRangeQuery<K, V> withUpperKeyBound(final K upper);

  /**
   * Interactive scan query that returns all records in the store. * For each key the records valid
   * @paramwithin <K>the specified time Therange keyare type
returned. * In *case @paramthe <V>time range is Thenot valuespecified typejust
   */
 the publiclatest staticrecord <K,for V>each MultiVersionedRangeQuery<K,key V> withUpperKeyBound(final K upper);

  /**
   * Interactive scan query that returns all records in the store. * For each key the records validis returned.
   * @param <K> The key type
   * @param <V> The value type
   */
  public static <K, V> MultiVersionedRangeQuery<K, V> allKeys();

  /**
   * withinSpecifies the specifiedstarting time range are returned. * In case the time range is not specified just
   * the latest record for each key is returnedpoint for the key query. The range query returns all the records
   * that are valid in the time range starting from the timestamp {@code fromTime}.
   * @param <K>fromTime The starting keytime typepoint
   */
 @param <V> The value type
   */
  public static <K, V> MultiVersionedRangeQuery<K, V> allKeysfromTime(Instant fromTime);

  /**
   * Specifies the startingending time point for the key query. The range query returns all the records that
   * thathave are valid in the time range starting from the timestamp timestamp <= {@code fromTimestamptoTime}.
   * @param fromTimestamptoTime The startingending time point
   */
  public MultiVersionedRangeQuery<K, V> fromtoTime(Instant fromTimestamptoTime);

  /**
   * SpecifiesThe thelower endingbound timeof pointthe forquery, theif keyspecified.
 query. The range*/
 query returnspublic all the records thatOptional<K> lowerKeyBound();

  /**
   * haveThe timestampupper <=bound {@code asOfTimestamp}.
   * @param asOfTimestamp The ending time pointof the query, if specified
   */
  public MultiVersionedRangeQuery<K, V> asOf(Instant asOfTimestampOptional<K> upperKeyBound();


  /**
   * SpecifiesThe thestarting overalltime orderpoint of returnedthe recordsquery, byif timestampspecified
   */
  public MultiVersionedRangeQuery<K, V> orderByTimestampOptional<Instant> fromTime();

  /**
   * The Specifiesending thetime orderpoint of keys as descending.the query, if specified
   */
  public MultiVersionedRangeQuery<K, V> withDescendingKeysOptional<Instant> toTime();


  /**    
   * Specifies the order of the timestamps as descending.
   */
  public VersionedRangeQuery<K, V> withDescendingTimestamps();

  /**
   * The lower bound of the query, if specified.
   */
  public Optional<K> lowerKeyBound();

  /**
   * The upper bound of the query, if specified
   */
  public Optional<K> upperKeyBound();

  /**
   * The starting time point of the query, if specified
   */
  public Optional<Instant> fromTimestamp();

  /**
   * The ending time point of the query, if specified
   */
  public Optional<Instant> asOfTimestamp();

  /**
   * @return true if the query orders the returned records by key 
   */
  public boolean isOrderedByKey();

  /**
   * @return true if the query returns records in ascending order of keys
   */
  public boolean isKeyAscending();

  /**
   * @return true if the query returns records in ascending order of timestamps
   */
  public boolean isRangeAscending();

}

}

// example 1: MultiVersionedRangeQuery without specifying any time bound will be interpreted as all versions
final MultiVersionedRangeQuery<Integer, Integer> query1 =
        MultiVersionedRangeQuery.withKeyRange(1, 2);
final StateQueryRequest<KeyValueIterator<Integer, VersionedRecord<Integer>>> request1 =
        inStore("my_store").withQuery(query1);
 
final StateQueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>> versionedRangeResult1 = kafkaStreams.query(request1);
 
// Get the results from all partitions.
final Map<Integer, QueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>>> partitionResults = versionedRangeResult.getPartitionResults();
for (final Entry<Integer, QueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>>> entry : partitionResults.entrySet()) {
    try (final KeyValueIterator<Integer, VersionedRecord<Integer>> iterator = entry.getValue().getResult()) {

package org.apache.kafka.streams.state;
 
public interface VersionedKeyValueStore<K, V> extends StateStore {
      /**
     * Get the record associated with this key as of the specified timestamp (i.e.,
     * the existing record with the largest timestamp not exceeding the provided
     * timestamp bound).
     *
     * @param lowerKeyBound           The key that specifies the lower key bound of the range
     * @param upperKeyBound           The key that specifies the upper key bound of the range
 
     * @param fromTimestamp The timestamp lower bound. The records that have been inserted at
                            or before this timestamp and did not become tombstone at or before
                   while (iterator.hasNext()) {
       this timestamp will befinal retrievedKeyValue<Integer, andVersionedRecord<Integer>> returned.
record   = iterator.next();
  * @param asOfTimestamp The timestamp bound. This bound isInteger inclusive; if akey = record.key;
     *     Integer value = record.value.value();
          Long timestamp =  (for the specified key) exists with this timestamp, then
     *                      this is the record that will be returned.record.value.timestamp();
		  Long validTo = record.value.validTo();
	      System.out.println ("key,value: " + key + "," +value + ", timestamp: " + Instant.ofEpochSecond(timestamp)+ ", valid till: " + Instant.ofEpochSecond(validTo));  	
        }
     }
}
/* the @returnprinted Theoutput valuewill andbe
 timestamp (along with the validTokey,value: 1,1, timestamp) of the records with keys within the specified key range: 2023-01-01T10:00:00.00Z, valid till: 2023-01-05T10:00:00.00Z  
     *         as of the provided timestamp, or {@code null} if no such record exists
     *         (including if the provided timestamp bound is older than this store's history
     *         retention time, i.e., the store no longer contains data for the provided
     *         timestamp).
     * @throws NullPointerException       If null is used for lowerKeyBound or upperKeyBound.
     * @throws InvalidStateStoreException if the store is not initialized
     */
    VersionedRecord<V> get(K lowerKeyBound, K upperKeyBound, long fromTimestamp, long asOfTimestamp);
}