class RewrittenAnalysis implements ImmutableAnalysis {

  @Override
  public Optional<WindowExpression> getWindowExpression() {
    final Optional<WindowExpression> windowExpression = original.getWindowExpression();
    final Optional<RefinementInfo> refinementInfo = original.getRefinementInfo();

    /* Return the original window expression, unless there is no grace period provided during a
    suppression, in which case we rewrite the window expression to have a default grace period
    of zero.
    */
    if (!(windowExpression.isPresent()
        && !windowExpression.get().getKsqlWindowExpression().getGracePeriod().isPresent()
        && refinementInfo.isPresent()
        && refinementInfo.get().getOutputRefinement() == OutputRefinement.FINAL
         )
    ) {
      return original.getWindowExpression();
    }
    final WindowExpression window = original.getWindowExpression().get();

    final KsqlWindowExpression ksqlWindowNew;
    final KsqlWindowExpression ksqlWindowOld = window.getKsqlWindowExpression();

    final Optional<NodeLocation> location = ksqlWindowOld.getLocation();
    final Optional<WindowTimeClause> retention = ksqlWindowOld.getRetention();

    if (ksqlWindowOld instanceof HoppingWindowExpression) {
      ksqlWindowNew = new HoppingWindowExpression(
          location,
          ((HoppingWindowExpression) ksqlWindowOld).getSize(),
          ((HoppingWindowExpression) ksqlWindowOld).getAdvanceBy(),
          retention,
          Optional.of(zeroGracePeriod)
      );
    } else if (ksqlWindowOld instanceof TumblingWindowExpression) {
      ksqlWindowNew = new TumblingWindowExpression(
          location,
          ((TumblingWindowExpression) ksqlWindowOld).getSize(),
          retention,
          Optional.of(zeroGracePeriod)
      );
    } else if (ksqlWindowOld instanceof SessionWindowExpression) {
      ksqlWindowNew = new SessionWindowExpression(
          location,
          ((SessionWindowExpression) ksqlWindowOld).getGap(),
          retention,
          Optional.of(zeroGracePeriod)
      );
    } else {
      throw new KsqlException("WINDOW type must be HOPPING, TUMBLING, or SESSION");
    }
    return Optional.of(new WindowExpression(
        original.getWindowExpression().get().getWindowName(),
        ksqlWindowNew
    ));
  }

  public  RewrittenAnalysis(
      final ImmutableAnalysis original,
      final BiFunction<Expression, Context<Void>, Optional<Expression>> rewriter
  );

}