plan for a projection with case

projection-with-case-plan

{
	"@type": "ksqlPlanV1",
	"statementText": "CREATE STREAM S1 AS SELECT (CASE WHEN (ORDERS.ORDERUNITS < 2.0) THEN 'small' WHEN (ORDERS.ORDERUNITS < 4.0) THEN 'medium' ELSE 'large' END) CASE_RESAULT\nFROM ORDERS ORDERS\nEMIT CHANGES",
	"ddlCommand": {
		"@type": "createStreamV1",
		"sourceName": "S1",
		"schema": "`ROWKEY` STRING KEY, `CASE_RESAULT` STRING",
		"topicName": "S1",
		"formats": {
			"keyFormat": {
				"format": "KAFKA"
			},
			"valueFormat": {
				"format": "JSON"
			},
			"options": []
		}
	},
	"queryPlan": {
		"sources": ["ORDERS"],
		"sink": "S1",
		"physicalPlan": {
			"@type": "streamSinkV1",
			"properties": {
				"queryContext": "S1"
			},
			"source": {
				"@type": "streamSelectV1",
				"properties": {
					"queryContext": "Project"
				},
				"source": {
					"@type": "streamSourceV1",
					"properties": {
						"queryContext": "KsqlTopic/Source"
					},
					"topicName": "test_topic",
					"formats": {
						"keyFormat": {
							"format": "KAFKA"
						},
						"valueFormat": {
							"format": "JSON"
						},
						"options": []
					},
					"sourceSchema": "`ROWKEY` STRING KEY, `ORDERUNITS` DOUBLE",
					"alias": "ORDERS"
				},
				"selectExpressions": ["(CASE WHEN (ORDERS.ORDERUNITS < 2.0) THEN 'small' WHEN (ORDERS.ORDERUNITS < 4.0) THEN 'medium' ELSE 'large' END) AS CASE_RESAULT"]
			},
			"formats": {
				"keyFormat": {
					"format": "KAFKA"
				},
				"valueFormat": {
					"format": "JSON"
				},
				"options": []
			},
			"topicName": "S1"
		},
		"queryId": "CSAS_S1_0"
	}
}