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

    // we only need to rewrite if we have a window expression and if we use emit final
    if (!windowExpression.isPresent()
        || !refinementInfo.isPresent()
        || refinementInfo.get().getOutputRefinement() == OutputRefinement.CHANGES) {
      return original.getWindowExpression();
    }

    final Optional<WindowTimeClause> gracePeriod;
    if (!windowExpression.get().getKsqlWindowExpression().getGracePeriod().isPresent()) {
      gracePeriod = Optional.of(zeroGracePeriod);
    } else {
      gracePeriod = windowExpression.get().getKsqlWindowExpression().getGracePeriod();
    }

    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,
          gracePeriod,
          Optional.of(OutputRefinement.FINAL)
      );
    } else if (ksqlWindowOld instanceof TumblingWindowExpression) {
      ksqlWindowNew = new TumblingWindowExpression(
          location,
          ((TumblingWindowExpression) ksqlWindowOld).getSize(),
          retention,
          gracePeriod,
          Optional.of(OutputRefinement.FINAL)
      );
    } else if (ksqlWindowOld instanceof SessionWindowExpression) {
      ksqlWindowNew = new SessionWindowExpression(
          location,
          ((SessionWindowExpression) ksqlWindowOld).getGap(),
          retention,
          gracePeriod,
          Optional.of(OutputRefinement.FINAL)
      );
    } else {
      throw new KsqlException("WINDOW type must be HOPPING, TUMBLING, or SESSION");
    }
    return Optional.of(new WindowExpression(
        original.getWindowExpression().get().getWindowName(),
        ksqlWindowNew
    ));
  }
@Test
  public void shouldCreateNewTumblingWindowWithZeroGracePeriodDefault() {
    // Given:
    when(windowExpression.getKsqlWindowExpression()).thenReturn(tumblingWindow);
    when(tumblingWindow.getGracePeriod()).thenReturn(gracePeriodOptional);
    when(gracePeriodOptional.isPresent()).thenReturn(false);
    when(tumblingWindow.getLocation()).thenReturn(location);
    when(tumblingWindow.getRetention()).thenReturn(retention);
    when(tumblingWindow.getSize()).thenReturn(size);
    when(windowExpression.getWindowName()).thenReturn(windowName);


    // When:
    Optional<WindowExpression> result = rewrittenAnalysis.getWindowExpression();

    // Then:
    assertThat(result.get().getKsqlWindowExpression().getGracePeriod(), not(Optional.empty()));
  }