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 finalMultiVersionedRangeQuery(
  boolean isKeyAscending;
  private final booleanOptional<K> isTimeAscending;
lower,
     private final boolean isOrderedByKey;
Optional<K> upper,
  private MultiVersionedRangeQuery(
      final Optional<K>Optional<Instant> lowerfromTime,
      final Optional<Instant> Optional<K>toTime) upper,{
    this.lower  final Optional<Instant> fromTimestamp,= lower;
    this.upper  final Optional<Instant> asOfTimestamp,= upper;
    this.fromTime  final boolean isOrderedByKey,= fromTime;
    this.toTime = finaltoTime;
 boolean isKeyAscending,}

   /** 
  final boolean* isTimeAscending)Interactive {
range query using a this.lower and upper =bound lower;
to filter the keys this.upper = upper;returned. * For each 
   * this.fromTimestampkey =the fromTimestamp;
records valid within the this.asOfTimestampspecified =time asOfTimestamp;
range are   thisreturned.isOrderedByKey =* isOrderedByKey;
In case the time this.isKeyAscendingrange =is isKeyAscending;
   * this.isTimeAscendingnot =specified isTimeAscending;
just the }

  /**latest record for each key is returned. 
   * Interactive@param rangelower queryThe usingkey athat lowerspecifies andthe upperlower bound of tothe filterrange the
 keys returned. * For@param each
upper The key *that keyspecifies the recordsupper validbound withinof the specifiedrange time
 range are returned. * In@param case<K> theThe timekey rangetype is
   * 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, final bound of the rangeK upper);


   /**
   * @paramInteractive upperrange Thequery keyusing thata specifieslower thebound upperto boundfilter of the range
keys returned.  * @param <K>   TheFor each key typethe
   * @paramrecords <V>valid within the Thespecified valuetime type
range are returned. */
 In publiccase staticthe <K,time V>range MultiVersionedRangeQuery<K,is V>not
 withRange(final K lower,* finalspecified K upper);

  /**
   * Interactive range query using a lower bound to filter the keys returned. * For each key thejust the latest record for each key is returned.
   * @param lower The key that specifies the lower bound of the range
   * records@param valid<K> within the specifiedThe timekey rangetype
 are returned. * In@param case<V> the time rangeThe isvalue nottype
   */
 specified justpublic thestatic latest<K, recordV> forMultiVersionedRangeQuery<K, eachV> keywithLowerKeyBound(final is returned.K lower);

  /**
 * @param lower* TheInteractive keyrange thatquery specifiesusing thea lower bound to offilter the range
keys returned.  * @paramFor <K>  each The key typethe
   * @paramrecords <V>valid within the Thespecified valuetime type
range are returned. */
 In publiccase staticthe <K,time V>range MultiVersionedRangeQuery<K,is V>not
 withLowerBound(final K lower);

*  /**specified just the latest record for each key is returned.
   * @param upper InteractiveThe rangekey querythat usingspecifies athe lower bound toof filterthe therange
 keys returned. * For each@param <K>   The key thetype
   * records@param valid<V> within the specifiedThe timevalue rangetype
 are returned. * In case the time range is not
   * specified just the latest record for each key is returned.
   * @param upper The key that specifies the lower bound of the range
   * @param <K>   The key type/
  public static <K, V> MultiVersionedRangeQuery<K, V> withUpperKeyBound(final K upper);

  /**
   * Interactive scan query that returns all records in the store. * For each key the records valid
   * within the specified time range are returned. * In case the time range is not specified just
   * @paramthe <V>latest record for The valueeach key is returned.
   * @param <K> The key type
   * @param <V> The value type
   */
  public static <K, V> MultiVersionedRangeQuery<K, V> withUpperBoundallKeys(final K upper);

  /**
   * InteractiveSpecifies scanthe querystarting thattime returnspoint allfor recordsthe inkey the storequery. The *range Forquery eachreturns keyall the records valid
   * within the specified time range that are returned.valid *in In case the time range starting from isthe nottimestamp specified just{@code fromTime}.
   * the@param latestfromTime recordThe forstarting each key is returned.
   * @param <K> The key typetime point
   * @param <V> The value type
   */
  public static <K, V> MultiVersionedRangeQuery<K, V> withNoBoundsfromTime(Instant fromTime);

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

  /**
   * SpecifiesThe thelower endingbound timeof pointthe forquery, theif key queryspecified.
 The range query*/
 returns allpublic the records thatOptional<K> lowerKeyBound();

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

  /**
   * Specifies theThe starting andtime endingpoint points ofof the range query, asif MINspecified
 and MAX respectively.*/
  public * Therefore, the query returns all the existing records in the state store with keys within theOptional<Instant> fromTime();

  /**
   * The ending time point of the query, if specified
   * specified key range./
  public * @throws RuntimeException if {@code fromTimestamp} or {@code asOfTimestamp} have been already
   *       Optional<Instant> toTime();

}

// 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()) {
        while (iterator.hasNext()) {
          final KeyValue<Integer, VersionedRecord<Integer>> record = iterator.next();
          Integer key = record.key;
      specified.
   */
   publicInteger MultiVersionedRangeQuery<K,value V> allVersions= record.value.value();

  /**
   * Specifies the overall order ofLong returnedtimestamp records by timestamp
   */
  public MultiVersionedRangeQuery<K, V> orderByTimestamp= record.value.timestamp();
		  Long validTo = record.value.validTo();

	   /**
   * Specifies the order of keys as descending.
   */
  public MultiVersionedRangeQuery<K, V> withDescendingKeys();

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

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

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

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

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

  /**
   * @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();

}