== Parsed Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@a631a1f
+- SubqueryAlias gif_count
+- Aggregate [window#81-T3000ms], [window#81-T3000ms AS window#69-T3000ms, count(1) AS count#80L]
+- Filter ((timestamp#1-T3000ms >= window#81-T3000ms.start) AND (timestamp#1-T3000ms < window#81-T3000ms.end))
+- Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(0 as bigint)) - cast(2 as bigint)) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion((((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(0 as bigint)) - cast(2 as bigint)) * 30000000) + 0) + 60000000), LongType, TimestampType)), logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(1 as bigint)) - cast(2 as bigint)) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion((((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(1 as bigint)) - cast(2 as bigint)) * 30000000) + 0) + 60000000), LongType, TimestampType)), logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49)], [window#81-T3000ms, logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49]
+- Filter Contains(url#32, gif)
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, split(logs#5, ,, -1)[6] AS bytes#49]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, split(logs#5, ,, -1)[5] AS responsecode#40]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, split(logs#5, ,, -1)[4] AS url#32]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, split(logs#5, ,, -1)[3] AS method#25]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, split(logs#5, ,, -1)[2] AS time#19]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, split(logs#5, ,, -1)[1] AS hostname#14]
+- Project [logs#5, timestamp#1-T3000ms, split(logs#5, ,, -1)[0] AS idx#10]
+- EventTimeWatermark timestamp#1: timestamp, 3 seconds
+- Project [logs#5, timestamp#1]
+- Generate explode(split(value#0, , -1)), false, [logs#5]
+- StreamingDataSourceV2Relation [value#0, timestamp#1], org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1@466d1a44, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 10, 33
== Analyzed Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@a631a1f
+- SubqueryAlias gif_count
+- Aggregate [window#81-T3000ms], [window#81-T3000ms AS window#69-T3000ms, count(1) AS count#80L]
+- Filter ((timestamp#1-T3000ms >= window#81-T3000ms.start) AND (timestamp#1-T3000ms < window#81-T3000ms.end))
+- Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(0 as bigint)) - cast(2 as bigint)) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion((((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(0 as bigint)) - cast(2 as bigint)) * 30000000) + 0) + 60000000), LongType, TimestampType)), logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(1 as bigint)) - cast(2 as bigint)) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion((((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) + cast(1 as bigint)) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / cast(30000000 as double))) END + cast(1 as bigint)) - cast(2 as bigint)) * 30000000) + 0) + 60000000), LongType, TimestampType)), logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49)], [window#81-T3000ms, logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, bytes#49]
+- Filter Contains(url#32, gif)
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, responsecode#40, split(logs#5, ,, -1)[6] AS bytes#49]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, url#32, split(logs#5, ,, -1)[5] AS responsecode#40]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, method#25, split(logs#5, ,, -1)[4] AS url#32]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, time#19, split(logs#5, ,, -1)[3] AS method#25]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, hostname#14, split(logs#5, ,, -1)[2] AS time#19]
+- Project [logs#5, timestamp#1-T3000ms, idx#10, split(logs#5, ,, -1)[1] AS hostname#14]
+- Project [logs#5, timestamp#1-T3000ms, split(logs#5, ,, -1)[0] AS idx#10]
+- EventTimeWatermark timestamp#1: timestamp, 3 seconds
+- Project [logs#5, timestamp#1]
+- Generate explode(split(value#0, , -1)), false, [logs#5]
+- StreamingDataSourceV2Relation [value#0, timestamp#1], org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1@466d1a44, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 10, 33
== Optimized Logical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@a631a1f
+- Aggregate [window#81-T3000ms], [window#81-T3000ms AS window#69-T3000ms, count(1) AS count#80L]
+- Project [window#81-T3000ms]
+- Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#81-T3000ms)) AND (timestamp#1-T3000ms >= window#81-T3000ms.start)) AND (timestamp#1-T3000ms < window#81-T3000ms.end))
+- Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms)], [window#81-T3000ms, timestamp#1-T3000ms]
+- EventTimeWatermark timestamp#1: timestamp, 3 seconds
+- Project [timestamp#1]
+- Filter Contains(split(logs#5, ,, -1)[4], gif)
+- Generate explode(split(value#0, , -1)), [0], false, [logs#5]
+- StreamingDataSourceV2Relation [value#0, timestamp#1], org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1@466d1a44, TextSocketV2[host: stream-emulator.data-science-tools.svc.cluster.local, port: 5551], 10, 33
== Physical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite@a631a1f
+- *(6) HashAggregate(keys=[window#81-T3000ms], functions=[count(1)], output=[window#69-T3000ms, count#80L])
+- StateStoreSave [window#81-T3000ms], state info [ checkpoint = file:/tmp/temporary-44a37146-c22b-4332-803b-47058dc7d369/state, runId = 2ad6ec75-02a9-40e4-8f05-f0dd3dbf4b8d, opId = 0, ver = 2, numPartitions = 200], Update, 1735309969784, 2
+- *(5) HashAggregate(keys=[window#81-T3000ms], functions=[merge_count(1)], output=[window#81-T3000ms, count#102L])
+- StateStoreRestore [window#81-T3000ms], state info [ checkpoint = file:/tmp/temporary-44a37146-c22b-4332-803b-47058dc7d369/state, runId = 2ad6ec75-02a9-40e4-8f05-f0dd3dbf4b8d, opId = 0, ver = 2, numPartitions = 200], 2
+- *(4) HashAggregate(keys=[window#81-T3000ms], functions=[merge_count(1)], output=[window#81-T3000ms, count#102L])
+- Exchange hashpartitioning(window#81-T3000ms, 200), true, [id=#685]
+- *(3) HashAggregate(keys=[window#81-T3000ms], functions=[partial_count(1)], output=[window#81-T3000ms, count#102L])
+- *(3) Project [window#81-T3000ms]
+- *(3) Filter (((isnotnull(timestamp#1-T3000ms) AND isnotnull(window#81-T3000ms)) AND (timestamp#1-T3000ms >= window#81-T3000ms.start)) AND (timestamp#1-T3000ms < window#81-T3000ms.end))
+- *(3) Expand [ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 0) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms), ArrayBuffer(named_struct(start, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 0), LongType, TimestampType), end, precisetimestampconversion(((((CASE WHEN (cast(CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) as double) = (cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) THEN (CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) + 1) ELSE CEIL((cast((precisetimestampconversion(timestamp#1-T3000ms, TimestampType, LongType) - 0) as double) / 3.0E7)) END + 1) - 2) * 30000000) + 60000000), LongType, TimestampType)), timestamp#1-T3000ms)], [window#81-T3000ms, timestamp#1-T3000ms]
+- EventTimeWatermark timestamp#1: timestamp, 3 seconds
+- *(2) Project [timestamp#1]
+- *(2) Filter Contains(split(logs#5, ,, -1)[4], gif)
+- Generate explode(split(value#0, , -1)), [timestamp#1], false, [logs#5]
+- *(1) Project [value#0, timestamp#1]
+- MicroBatchScan[value#0, timestamp#1] class org.apache.spark.sql.execution.streaming.sources.TextSocketTable$$anon$1
