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;

 final privateOptional<K> MultiVersionedRangeQuery(upper,
      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 lower;
upper bound to filter this.upperthe = upper;
    this.fromTimestamp = fromTimestamp;keys returned. * For each 
   * this.asOfTimestampkey =the asOfTimestamp;
records valid within the this.isOrderedByKeyspecified =time isOrderedByKey;
range are   thisreturned.isKeyAscending =* isKeyAscending;
In case the time this.isTimeAscendingrange =is isTimeAscending;
  }

  /**
 not specified *just Interactivethe rangelatest queryrecord usingfor aeach lowerkey andis 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.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 lower<V> The keyvalue thattype specifies
 the lower bound of the range*/ 
   *public @paramstatic upper<K, The key that specifies the upper bound of the rangeV> MultiVersionedRangeQuery<K, V> withKeyRange(final K lower, final K upper);


   /**
   * @paramInteractive <K>range query using Thea keylower type
bound to filter *the @paramkeys <V>returned. * For Theeach valuekey typethe
   */
  public static <K, V> MultiVersionedRangeQuery<K, V> withRange(final K lower, final K upper);

  /** records valid within the specified time range are returned. * In case the time range is not
   * Interactivespecified rangejust querythe usinglatest arecord lowerfor boundeach tokey filteris thereturned.
 keys returned. * @param Forlower eachThe key the
that specifies the *lower recordsbound validof within the specified time range
 are returned. * In@param case<K> the time rangeThe iskey nottype
   * specified@param just<V> the latest recordThe forvalue eachtype
 key is returned.*/
  public *static @param<K, lowerV> TheMultiVersionedRangeQuery<K, keyV> that specifies thewithLowerKeyBound(final K lower);

 bound of the range /**
   * @paramInteractive <K>range query using Thea keylower type
bound to filter *the @paramkeys <V>returned. * For Theeach valuekey typethe
   */
 records publicvalid staticwithin <K,the V>specified MultiVersionedRangeQuery<K,time V>range withLowerBound(final K lower);

  /**
   * Interactive range query using a lower bound to filter the keys returned. * For each key theare returned. * In case the time range is not
   * specified just the latest record for each key is returned.
   * records@param validupper withinThe thekey specifiedthat timespecifies rangethe arelower returned. * In case the time range is notbound of the range
   * specified just the latest record for each key is returned.
   * @param upper @param <K>   The key that specifies the lower bound of the range
   * @param <K>   The key type
   * @param <V>   The value type
   */
  public static <K, V> MultiVersionedRangeQuery<K, V> withUpperBoundwithUpperKeyBound(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
   * the latest record for each key is returned.
   * @param <K> The key type
   * @param <V> The value type
   */
  public static <K, V> MultiVersionedRangeQuery<K, V> withNoBoundsallKeys();

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

  /**
   * Specifies the ending time point for the key query. The range query returns all the records that
   * have timestamp <= {@code asOfTimestamptoTime}.
   * @param asOfTimestamptoTime The ending time point
   */
  public MultiVersionedRangeQuery<K, V> asOftoTime(Instant asOfTimestamptoTime);

  /**
   * SpecifiesThe thelower starting and ending points bound of the range query, as MIN and MAX respectivelyif specified.
   */
 Therefore, thepublic query returns all the existing records in the state store with keys within theOptional<K> lowerKeyBound();

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

   /**
 @throws RuntimeException if* {@codeThe fromTimestamp}starting ortime {@code asOfTimestamp} have been alreadypoint of the query, if specified
   */
  public Optional<Instant> fromTime();

  /**
   * The ending time point of the query,          specified.if specified
   */
   public MultiVersionedRangeQuery<K, V> allVersionsOptional<Instant> toTime();

}
  /**
   * Specifies the overall order of returned records by timestamp
   */
  public MultiVersionedRangeQuery<K, V> orderByTimestamp();

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

}

// 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;
          Integer value = record.value.value();
          Long timestamp = 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 printed output will be
    key,value: 1,1, timestamp: 2023-01-01T10:00:00.00Z, valid till: 2023-01-05T10:00:00.00Z  
    key,value: 1,2, timestamp: 2023-01-15T10:00:00.00Z, valid till: 2023-01-20T10:00:00.00Z 
    key,value: 1,3, timestamp: 2023-01-20T10:00:00.00Z, valid till: now 
    key,value: 2,20, timestamp: 2023-01-10T10:00:00.00Z, valid till: 2023-01-25T10:00:00.00Z 
    key,value: 2,30, timestamp: 2023-01-25T10:00:00.00Z, valid till: now 
*/

// example 2: The value of the records with key range (1,2) from 2023-01-17 Time: 10:00:00.00Z till 2023-01-30 T10:00:00.00Z


MultiVersionedRangeQuery<Integer, Integer> query2 = MultiVersionedRangeQuery.withKeyRange(1, 2);
query2 = query2.fromTime(Instant.parse(2023-01-17T10:00:00.00Z)).toTime(Instant.parse(2023-01-30T10:00:00.00Z));
 
final StateQueryRequest<KeyValueIterator<Integer, VersionedRecord<Integer>>> request2 =
        inStore("my_store").withQuery(query2);
 
final StateQueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>> versionedRangeResult2 = kafkaStreams.query(request2);
 
// Get the results from all partitions.
final Map<Integer, QueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>>> partitionResults2 = versionedRangeResult2.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;
          Integer value = record.value.value();
          Long timestamp = 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 printed output will be   
    key, value: 2,30, timestamp: 2023-01-25T10:00:00.00Z, valid till: now 
    key, value: 1,3, timestamp: 2023-01-20T10:00:00.00Z, valid till: now
    key, value: 1,2, timestamp: 2023-01-15T10:00:00.00Z, valid till: 2023-01-20T10:00:00.00Z
    key, value: 2,20, timestamp: 2023-01-10T10:00:00.00Z, valid till: 2023-01-25T10:00:00.00Z 
*/

final MultiVersionedRangeQuery<Integer, Integer> query =
        MultiVersionedRangeQuery.withKeyRange(1, 2).from(Instant.parse(2023-08-03T10:37:30.00Z)).asOf(Instant.parse(2023-09-04T10:37:30.00Z));
 
final StateQueryRequest<KeyValueIterator<Integer, VersionedRecord<Integer>>> request =
        inStore("my_store").withQuery(query);
 
final StateQueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>> versionedRangeResult = kafkaStreams.query(request);
 
// 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;
          Integer value = record.value.value();
          Long timestamp = record.value.timestamp();
        }
     }
}