PARTITION BY not working when using a column alias in CREATE STREAM

[turbo-ele]

Describe the bug

When trying to create a stream from an existing stream and try to add a PARTITION BY with an alias i get the following message:

Invalid identifier for PARTITION BY clause: 'MY_ALIAS' Only columns from the source schema can be referenced in the PARTITION BY clause.

So the following create stream statement now fails:

CREATE STREAM bar
      WITH (
        VALUE_FORMAT='JSON'
      ) AS
      SELECT
          obj->prop AS my_alias
      FROM
          foo
      PARTITION BY my_alias;

This was working in confluent 5.2.1 (sorry i skipped some versions, but also couldn't find anything related in the release notes or issues)

To Reproduce
Steps to reproduce the behavior, include:

1. The version of KSQL: CLI v5.5.0, Server v5.5.0 using docker-image confluentinc/cp-ksqldb-cli:5.5.0

2. Any SQL statements you ran

CREATE STREAM foo (obj struct<prop int>) 
WITH (kafka_topic='foo',value_format='JSON',partitions=1);

CREATE STREAM bar
      WITH (
        VALUE_FORMAT='JSON'
      ) AS
      SELECT
          obj->prop AS my_alias
      FROM
          foo
      PARTITION BY my_alias;

It will only work when not using an alias:

CREATE STREAM bar
      WITH (
        VALUE_FORMAT='JSON'
      ) AS
      SELECT
          obj->prop
      FROM
          foo
      PARTITION BY obj->prop;

Additional context
Workarounds like proposed here(#2200) are not working anymore.

[agavra]

Hello @turbo-ele thanks for reporting this! - I think #2701 is the ticket that you're looking for that describes this breaking change (and you can see further discussion on the PR that implemented the change #3982). Let me know if that helps.

[turbo-ele]

Thanks @agavra for the clarification and the reference.
Sorry for the noise