@Override
  public SchemaKStream<?> buildStream(final PlanBuildContext buildContext) {
    final QueryContext.Stacker contextStacker = buildContext.buildNodeContext(getId().toString());
    final SchemaKStream<?> schemaKStream = getSource().buildStream(buildContext);

    if (!(schemaKStream instanceof SchemaKTable)) {
      throw new KsqlException("Failed in suppress node. Expected to find a Table, but "
          + "found a stream instead.");
    }

    return (((SchemaKTable<?>) schemaKStream)
        .suppress(
            refinementInfo,
            valueFormat.getFormatInfo(),
            contextStacker
        ));
  }
@Test
  public void shouldThrowOnSuppressOnStream() {

    // Given:
    when(sourceNode.getNodeOutputType()).thenReturn(DataSourceType.KSTREAM);
    node = new SuppressNode(NODE_ID, sourceNode, refinementInfo);

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> node.buildStream(planBuildContext)
    );

    // Then
    assertThat(e.getMessage(), containsString("Failed in suppress node. Expected to find a Table, but found a stream instead"));
  }