plan for average aggregation

aggregate-plan

{
	"@type": "ksqlPlanV1",
	"statementText": "CREATE TABLE AVG AS SELECT\n  TEST.ID ID,\n  (SUM(TEST.VALUE) / COUNT(TEST.ID)) AVG\nFROM TEST TEST\nGROUP BY TEST.ID\nHAVING ((SUM(TEST.VALUE) / COUNT(TEST.ID)) > 25)\nEMIT CHANGES",
	"ddlCommand": {
		"@type": "createTableV1",
		"sourceName": "AVG",
		"schema": "`ROWKEY` STRING KEY, `ID` BIGINT, `AVG` BIGINT",
		"keyField": "ID",
		"topicName": "AVG",
		"formats": {
			"keyFormat": {
				"format": "KAFKA"
			},
			"valueFormat": {
				"format": "DELIMITED"
			},
			"options": []
		}
	},
	"queryPlan": {
		"sources": ["TEST"],
		"sink": "AVG",
		"physicalPlan": {
			"@type": "tableSinkV1",
			"properties": {
				"queryContext": "AVG"
			},
			"source": {
				"@type": "tableSelectV1",
				"properties": {
					"queryContext": "Aggregate/Project"
				},
				"source": {
					"@type": "tableFilterV1",
					"properties": {
						"queryContext": "Aggregate/HavingFilter"
					},
					"source": {
						"@type": "streamAggregateV1",
						"properties": {
							"queryContext": "Aggregate/Aggregate"
						},
						"source": {
							"@type": "streamGroupByKeyV1",
							"properties": {
								"queryContext": "Aggregate/GroupBy"
							},
							"source": {
								"@type": "streamSelectV1",
								"properties": {
									"queryContext": "Aggregate/Prepare"
								},
								"source": {
									"@type": "streamSourceV1",
									"properties": {
										"queryContext": "KsqlTopic/Source"
									},
									"topicName": "test_topic",
									"formats": {
										"keyFormat": {
											"format": "KAFKA"
										},
										"valueFormat": {
											"format": "DELIMITED"
										},
										"options": []
									},
									"sourceSchema": "`ROWKEY` STRING KEY, `ID` BIGINT, `NAME` STRING, `VALUE` BIGINT",
									"alias": "TEST"
								},
								"selectExpressions": ["TEST.ID AS KSQL_INTERNAL_COL_0", "TEST.VALUE AS KSQL_INTERNAL_COL_1"]
							},
							"internalFormats": {
								"keyFormat": {
									"format": "KAFKA"
								},
								"valueFormat": {
									"format": "DELIMITED"
								},
								"options": []
							}
						},
						"internalFormats": {
							"keyFormat": {
								"format": "KAFKA"
							},
							"valueFormat": {
								"format": "DELIMITED"
							},
							"options": []
						},
						"nonAggregateColumns": ["KSQL_INTERNAL_COL_0", "KSQL_INTERNAL_COL_1"],
						"aggregationFunctions": ["SUM(KSQL_INTERNAL_COL_1)", "COUNT(KSQL_INTERNAL_COL_0)", "SUM(KSQL_INTERNAL_COL_1)", "COUNT(KSQL_INTERNAL_COL_0)"]
					},
					"filterExpression": "((KSQL_AGG_VARIABLE_2 / KSQL_AGG_VARIABLE_3) > 25)"
				},
				"selectExpressions": ["KSQL_INTERNAL_COL_0 AS ID", "(KSQL_AGG_VARIABLE_0 / KSQL_AGG_VARIABLE_1) AS AVG"]
			},
			"formats": {
				"keyFormat": {
					"format": "KAFKA"
				},
				"valueFormat": {
					"format": "DELIMITED"
				},
				"options": []
			},
			"topicName": "AVG"
		},
		"queryId": "CTAS_AVG_0"
	}
}