notebooks/03-query/13-functions-window.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "ad4e60b6",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Using window functions\n",
    "\n",
    "<!--\n",
    "  ~ Licensed to the Apache Software Foundation (ASF) under one\n",
    "  ~ or more contributor license agreements.  See the NOTICE file\n",
    "  ~ distributed with this work for additional information\n",
    "  ~ regarding copyright ownership.  The ASF licenses this file\n",
    "  ~ to you under the Apache License, Version 2.0 (the\n",
    "  ~ \"License\"); you may not use this file except in compliance\n",
    "  ~ with the License.  You may obtain a copy of the License at\n",
    "  ~\n",
    "  ~   http://www.apache.org/licenses/LICENSE-2.0\n",
    "  ~\n",
    "  ~ Unless required by applicable law or agreed to in writing,\n",
    "  ~ software distributed under the License is distributed on an\n",
    "  ~ \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n",
    "  ~ KIND, either express or implied.  See the License for the\n",
    "  ~ specific language governing permissions and limitations\n",
    "  ~ under the License.\n",
    "  -->\n",
    "\n",
    "A [window function](https://druid.apache.org/docs/latest/querying/sql-window-functions/#window-function-reference) operates over a \"window\" of rows that you specify, and then emits a value to each row of query results.\n",
    "\n",
    "This tutorial demonstrates how to define windows, and gives examples of window functions in action."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8d6bbbcb",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Prerequisites\n",
    "\n",
    "This tutorial works with Druid 31 or later.\n",
    "\n",
    "Launch this tutorial and all prerequisites using the `druid-jupyter` profile of the Docker Compose file for Jupyter-based Druid tutorials. For more information, see the Learn Druid repository [readme](https://github.com/implydata/learn-druid)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "56e7e84e-37b5-4e0c-9135-70a062f0098c",
   "metadata": {},
   "source": [
    "## Initialization\n",
    "\n",
    "The following cells set up the notebook and learning environment ready for use."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d17919fd-0a44-4938-a2f9-5c447d4d4834",
   "metadata": {},
   "source": [
    "### Set up a connection to Apache Druid\n",
    "\n",
    "Run the next cell to set up the Druid Python client's connection to Apache Druid.\n",
    "\n",
    "If successful, the Druid version number will be shown in the output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b7f08a52",
   "metadata": {},
   "outputs": [],
   "source": [
    "import druidapi\n",
    "import os\n",
    "\n",
    "druid_headers = {'Content-Type': 'application/json'}\n",
    "\n",
    "if 'DRUID_HOST' not in os.environ.keys():\n",
    "    druid_host=f\"http://localhost:8888\"\n",
    "else:\n",
    "    druid_host=f\"http://{os.environ['DRUID_HOST']}:8888\"\n",
    "\n",
    "print(f\"Opening a connection to {druid_host}.\")\n",
    "druid = druidapi.jupyter_client(druid_host)\n",
    "display = druid.display\n",
    "sql_client = druid.sql\n",
    "status_client = druid.status\n",
    "\n",
    "status_client.version"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3b55af57-9c79-4e45-a22c-438c1b94112e",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Create a table using batch ingestion\n",
    "\n",
    "Run the following cell to create a table using batch ingestion. Only the channel, user, and delta columns are ingested.\n",
    "\n",
    "When completed, you'll see a description of the final table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "959e3c9b",
   "metadata": {},
   "outputs": [],
   "source": [
    "table_name = 'example-wikipedia-windows'\n",
    "\n",
    "sql='''\n",
    "REPLACE INTO \"''' + table_name + '''\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"https://druid.apache.org/data/wikipedia.json.gz\"]}',\n",
    "    '{\"type\":\"json\"}'\n",
    "  )\n",
    ") EXTEND (\"timestamp\" VARCHAR, \"channel\" VARCHAR, \"user\" VARCHAR, \"delta\" BIGINT))\n",
    "SELECT\n",
    "  TIME_PARSE(\"timestamp\") AS \"__time\",\n",
    "  \"channel\",\n",
    "  \"user\",\n",
    "  \"delta\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY DAY\n",
    "'''\n",
    "\n",
    "display.run_task(sql)\n",
    "sql_client.wait_until_ready(table_name)\n",
    "display.table(table_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "42c51c97-b1b4-47df-97df-7acf8961b119",
   "metadata": {},
   "source": [
    "## Ordering windows\n",
    "\n",
    "In this first section, you will define a window of ordered rows to add ranks, number rows, and to bucket results."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a1efa147-a438-40d9-ad37-602d2d928f93",
   "metadata": {},
   "source": [
    "### Ranking results\n",
    "\n",
    "Run the following cell to produce an hourly activity timeline by channel, ordering them in descending order by the sum total number of edits."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7b71f7ee-4a07-4843-b217-8b4dcd8557f5",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT\n",
    "    TIME_FLOOR(__time, 'PT1H') as \"period\", \n",
    "    channel,\n",
    "    user,\n",
    "    SUM(delta) AS \"net_user_changes\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE channel IN ('#kk.wikipedia', '#lt.wikipedia')\n",
    "  AND TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "GROUP BY 1, 2, 3\n",
    "ORDER BY SUM(delta) DESC\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f8f5347c-0874-4590-bf12-ff8fb7772d1c",
   "metadata": {},
   "source": [
    "Review the query in the next cell.\n",
    "\n",
    "* The RANK window function adds a new dimension `rank_edits`.\n",
    "* An OVER clause defines the window to use with the window function.\n",
    "* The window is ordered by SUM(delta) descending.\n",
    "\n",
    "Run the cell to see the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d8e15cee-ef40-4123-a18d-2054115d2df5",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT\n",
    "    TIME_FLOOR(__time, 'PT1H') as \"period\", \n",
    "    channel,\n",
    "    user,\n",
    "    SUM(delta) AS \"net_user_changes\",\n",
    "    RANK() OVER (\n",
    "        ORDER BY SUM(delta) DESC\n",
    "        ) AS \"rank_edits\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE channel IN ('#kk.wikipedia', '#lt.wikipedia')\n",
    "  AND TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "GROUP BY 1, 2, 3\n",
    "ORDER BY 4 DESC\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "25a3489a-f446-4a26-bf21-5728681f3e5d",
   "metadata": {},
   "source": [
    "For each row, Druid has added a rank ordered by SUM(delta) descending.\n",
    "\n",
    "Run the next cell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0ec1bcb8-fdf9-4b61-a7a3-c8549aa42913",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT\n",
    "    TIME_FLOOR(__time, 'PT1H') as \"period\", \n",
    "    channel,\n",
    "    user,\n",
    "    SUM(delta) AS \"net_user_changes\",\n",
    "    RANK() OVER (\n",
    "        ORDER BY SUM(delta)\n",
    "        ) AS \"rank_edits\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE channel IN ('#kk.wikipedia', '#lt.wikipedia')\n",
    "  AND TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "GROUP BY 1, 2, 3\n",
    "ORDER BY 4 DESC\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9ff37a2f-ca08-43ca-a7e1-c9753d4550e7",
   "metadata": {},
   "source": [
    "Notice that the rank is now ordered by SUM(delta) ascending, while the results remain in descending order.\n",
    "\n",
    "In the next query, the outer ORDER BY has been changed so that the final result list is ordered by time.\n",
    "\n",
    "Run the next cell to see the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c87443df-0a75-4312-8d56-97d664633736",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT\n",
    "    TIME_FLOOR(__time, 'PT1H') as \"period\", \n",
    "    channel,\n",
    "    user,\n",
    "    SUM(delta) AS \"net_user_changes\",\n",
    "    RANK() OVER (\n",
    "        ORDER BY SUM(delta)\n",
    "        ) AS \"rank_edits\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE channel IN ('#kk.wikipedia', '#lt.wikipedia')\n",
    "  AND TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "GROUP BY 1, 2, 3\n",
    "ORDER BY 1\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "befd573b-8238-43c0-b4b4-df1c8a91bfc1",
   "metadata": {},
   "source": [
    "The `rank_edits` window remains in SUM(delta) ascending order, and the final results are ordered by `period`.\n",
    "\n",
    "In the next query:\n",
    "\n",
    "* `rank_editors` is a new column produced by an additional RANK function.\n",
    "* The window is ordered by COUNT(DISTINCT user) descending.\n",
    "\n",
    "Run the cell to see the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "71c7de02-57ff-497e-914f-583f57125522",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT\n",
    "    TIME_FLOOR(__time, 'PT1H') as \"period\", \n",
    "    RANK( ) OVER ( ORDER BY SUM(delta) ) AS \"rank_edits\",\n",
    "    RANK( ) OVER ( ORDER BY COUNT(DISTINCT user) DESC ) AS \"rank_editors\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE channel = '#en.wikipedia'\n",
    "AND TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "GROUP BY 1\n",
    "ORDER BY 1\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8f494135-8fe2-4e8e-a963-ad7cf68adb03",
   "metadata": {},
   "source": [
    "Notice that `rank_editors` is duplicated at position 10.\n",
    "\n",
    "You may want to:\n",
    "\n",
    "* Adjust the SQL above to include a COUNT(DISTINCT user) to find out why.\n",
    "* Change the window function to DENSE_RANK to see how this affects the results.\n",
    "* Add a PERCENT_RANK window function. Remember to defining a window using OVER and ORDER BY."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5b2e83ff-fb8a-453b-a7ff-8ef0e4b92af7",
   "metadata": {},
   "source": [
    "### Numerating and bucketing results\n",
    "\n",
    "The following query includes two more window functions:\n",
    "\n",
    "* ROW_NUMBER adds a monotonically increasing number.\n",
    "* NTILE splits the result rows into buckets of (near to) equal size, and gives each row a bucket number.\n",
    "\n",
    "The functions both operate over COUNT(DISTINCT user) in descending order.\n",
    "\n",
    "Run the cell to see the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "694900d0-891f-41bd-9b45-5ae957385244",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT \n",
    "    channel,\n",
    "    SUM(delta) AS \"edits\",\n",
    "    ROW_NUMBER() OVER (\n",
    "        ORDER BY COUNT(DISTINCT user) DESC\n",
    "        ) AS row_no,\n",
    "    NTILE(6) OVER (\n",
    "        ORDER BY COUNT(DISTINCT user) DESC\n",
    "        ) AS ntile_val\n",
    "FROM \"{table_name}\"\n",
    "WHERE TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "GROUP BY 1\n",
    "ORDER BY 3\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7cd8269f-dc22-4077-8500-90a47557494f",
   "metadata": {},
   "source": [
    "In the next query, a LIMIT is applied to the main query.\n",
    "\n",
    "What do you expect to happen?\n",
    "\n",
    "Run the cell to see the result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2858f3a1-37b3-4e18-8cc6-ac3ddd3ce4c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT \n",
    "    channel,\n",
    "    SUM(delta) AS \"edits\",\n",
    "    ROW_NUMBER() OVER (\n",
    "        ORDER BY COUNT(DISTINCT user) DESC\n",
    "        ) AS row_no,\n",
    "    NTILE(6) OVER (\n",
    "        ORDER BY COUNT(DISTINCT user) DESC\n",
    "        ) AS ntile_val\n",
    "FROM \"{table_name}\"\n",
    "WHERE TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "GROUP BY 1\n",
    "ORDER BY 2 DESC\n",
    "LIMIT 10\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "74368b02-8e0b-41d7-b46b-30da3a3076bd",
   "metadata": {},
   "source": [
    "## Partitioning windows\n",
    "\n",
    "Use PARTITION BY in window definitions to align window function calculations with a dimension value in each result row."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "edea4279-3433-43de-b0af-ba2d7f3b9798",
   "metadata": {},
   "source": [
    "Run the following cell to calcluate the SUM total edits in two specific channels in a day in the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "16e0a430-559e-4f23-8180-dd3b8a09fa8a",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT \n",
    "    channel,\n",
    "    SUM(delta) AS \"edits\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "AND channel IN ('#kk.wikipedia', '#lt.wikipedia')\n",
    "GROUP BY 1\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b874f28f-48b3-403e-82a9-11bf88dd3e55",
   "metadata": {},
   "source": [
    "Review the query below.\n",
    "\n",
    "* The main GROUP BY emits a sum of `edits` for each `channel` and `user`.\n",
    "* ROW_NUMBER and NTILE work as before.\n",
    "* The `channel_edits` column is new.\n",
    "  * The SUM() function is calculated over a window.\n",
    "  * The window is partitioned by `channel`.\n",
    "* Results are limited to a specific day and `channel`.\n",
    "\n",
    "Run the next cell to see the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9acf34ca-34cd-426c-b026-4dcd33c523c7",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT \n",
    "    channel,\n",
    "    user,\n",
    "    SUM(delta) AS \"edits\",\n",
    "    ROW_NUMBER() OVER (\n",
    "        ORDER BY COUNT(DISTINCT user) DESC\n",
    "        ) AS \"row_no\",\n",
    "    NTILE(6) OVER (\n",
    "        ORDER BY COUNT(DISTINCT user) DESC\n",
    "        ) AS \"ntile_val\",\n",
    "    SUM(SUM(delta)) OVER (\n",
    "        PARTITION BY channel\n",
    "        ) AS \"channel_edits\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "AND channel IN ('#kk.wikipedia', '#lt.wikipedia')\n",
    "GROUP BY 1, 2\n",
    "ORDER BY 1, 2\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3faa66a6-4e54-41f2-bd9a-bb0a5132b171",
   "metadata": {},
   "source": [
    "Notice how `channel_edits` matches the totals calculated earlier and added this to each matching row.\n",
    "\n",
    "The same technique can be used with a number of [standard aggregation functions](https://druid.apache.org/docs/latest/querying/sql-aggregations/). Run the next cell to see some examples."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6c56fed-b91b-40dd-90ad-ad675f4d68a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql=f\"\"\"\n",
    "SELECT user,\n",
    "    channel,\n",
    "    SUM(delta) AS \"edits\",\n",
    "    ROW_NUMBER() OVER (\n",
    "        ORDER BY COUNT(DISTINCT user) DESC\n",
    "        ) AS row_no,\n",
    "    NTILE(6) OVER (\n",
    "        ORDER BY COUNT(DISTINCT user) DESC\n",
    "        ) AS ntile_val,\n",
    "    SUM(SUM(delta)) OVER (\n",
    "        PARTITION BY channel\n",
    "        ) AS channel_edits,\n",
    "    SUM(SUM(delta)) OVER (\n",
    "        PARTITION BY user\n",
    "        ) AS user_edits,\n",
    "    COUNT(channel) OVER (\n",
    "        PARTITION BY user\n",
    "        ) AS user_row_count\n",
    "FROM \"{table_name}\"\n",
    "WHERE TIME_IN_INTERVAL(\"__time\",'2016-06-27/P1D')\n",
    "AND user LIKE 'L%'\n",
    "GROUP BY 1, 2\n",
    "HAVING SUM(delta) > 5000\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a65d68e4-893d-4027-8ca6-0f31b2cbf2b0",
   "metadata": {},
   "source": [
    "Notice that `Lsjbot` has made multiple edits across multiple channels in the same period."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "622f2158-75c9-4b12-bd8a-c92d30994c1f",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Ordering partitioned windows\n",
    "\n",
    "Use ORDER and PARTITION to order the rows inside each partitioned window.\n",
    "\n",
    "* The outer GROUP BY and SUM generates the `edits` for each `period`, `channel`, and `user`.\n",
    "* A RANK function is used to create `rank_within_channel_day`.\n",
    "\n",
    "Run the cell to see the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "20f1317f-a24a-4dc1-aaa0-7a7538c880e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT\n",
    "    TIME_FLOOR(__time, 'PT1H') as \"period\", \n",
    "    channel, \n",
    "    user,\n",
    "    SUM(delta) AS \"edits\",\n",
    "    RANK() OVER (\n",
    "        PARTITION BY channel\n",
    "        ORDER BY SUM(delta) DESC\n",
    "        ) AS \"rank_within_channel_day\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE channel IN ('#kk.wikipedia', '#lt.wikipedia')\n",
    "  AND TIME_IN_INTERVAL (\"__time\",'2016-06-27/P1D')\n",
    "GROUP BY 1, 2, 3\n",
    "ORDER BY 2, 5\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ad15b421-8bbf-436b-926c-26063dbadc02",
   "metadata": {},
   "source": [
    "Notice how the `rank_within_channel_day` has been calculated per channel, and is ordered by the number of edits in descending order."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "69e8ed88-2386-48c2-bc70-b299dd8ac016",
   "metadata": {},
   "source": [
    "## Limiting windows\n",
    "\n",
    "Run the following cell to produce a timeline of the number of edits in channels that begin with `#h`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9341c714-4e7d-4563-8b05-42557801c35d",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT\n",
    "    channel,\n",
    "    TIME_FLOOR(__time, 'PT1H') AS \"period\",\n",
    "    SUM(delta) AS \"edits\",\n",
    "FROM \"{table_name}\"\n",
    "WHERE \"channel\" LIKE '#h%'\n",
    "  AND TIME_IN_INTERVAL(\"__time\",'2016-06-27/PT12H')\n",
    "GROUP BY 1, 2\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c9badfce-cbfc-4395-bec5-8d52c003f090",
   "metadata": {},
   "source": [
    "Use ROWS BETWEEN to [limit the window](https://druid.apache.org/docs/latest/querying/sql-window-functions/#window-function-syntax) over which a function operates.\n",
    "\n",
    "In the following query, a new column `cumulative_channel_edits` has been added. It operates over a window that is:\n",
    "\n",
    "* Ordered by `__time`.\n",
    "* Partitioned by `channel`.\n",
    "* Scoped to include all preceding rows up to the current row.\n",
    "\n",
    "The result is a cumulative total, built up from all the results of the GROUP BY.\n",
    "\n",
    "A second new column, `rolling_max_edits` operates over a window that is:\n",
    "\n",
    "* Ordered by `__time`.\n",
    "* Partitioned by `channel`.\n",
    "* Scoped to the preceding 3 rows and the current row.\n",
    "\n",
    "This produces a channel-specific rolling maximum value over time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "59f6e587-5510-4de9-ab70-3a39dc356805",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT\n",
    "    channel,\n",
    "    TIME_FLOOR(__time, 'PT1H') AS \"period\",\n",
    "    SUM(delta) AS \"edits\",\n",
    "    SUM(SUM(delta)) OVER (\n",
    "        PARTITION BY channel\n",
    "        ORDER BY TIME_FLOOR(__time, 'PT1H')\n",
    "        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW\n",
    "        ) AS \"cumulative_channel_edits\",\n",
    "    MAX(SUM(delta)) OVER (\n",
    "        PARTITION BY channel\n",
    "        ORDER BY TIME_FLOOR(__time, 'PT1H')\n",
    "        ROWS BETWEEN 3 PRECEDING AND CURRENT ROW\n",
    "        ) AS \"rolling_max_edits\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE \"channel\" LIKE '#h%'\n",
    "  AND TIME_IN_INTERVAL(\"__time\",'2016-06-27/PT12H')\n",
    "GROUP BY 1, 2\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "75b5a98c-1016-4642-a2ad-e793d9772ea2",
   "metadata": {},
   "source": [
    "## Row-specific windows\n",
    "\n",
    "LAG, LEAD, and other functions address specific rows in the window that you define.\n",
    "\n",
    "Run the next cell to see the results of a query where:\n",
    "\n",
    "* `previous_edit_sum` returns the value of the previous row's SUM(delta).\n",
    "* `nextNext_edit_sum` returns the value of SUM(delta) two rows ahead.\n",
    "* `first_edit_sum` returns the very first SUM(delta) in the window.\n",
    "* `final_edit_sum` returns the very last SUM(delta) in the window."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "81b34b1d-c5ac-485a-83fc-f98373b7bfe6",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT TIME_FLOOR(\"__time\",'PT1H') AS \"period\",\n",
    "    channel,\n",
    "    SUM(delta) AS \"edit_sum\",\n",
    "    LAG(SUM(delta)) OVER (\n",
    "        PARTITION BY channel\n",
    "        ORDER BY TIME_FLOOR(\"__time\",'PT1H')\n",
    "        ) AS \"previous_edit_sum\",\n",
    "    LEAD(SUM(delta), 2) OVER (\n",
    "        PARTITION BY channel\n",
    "        ORDER BY TIME_FLOOR(\"__time\",'PT1H')\n",
    "        ) AS \"nextNext_edit_sum\",\n",
    "    FIRST_VALUE(SUM(delta)) OVER (\n",
    "        PARTITION BY channel\n",
    "        ) AS \"first_edit_sum\",\n",
    "    LAST_VALUE(SUM(delta)) OVER (\n",
    "        PARTITION BY channel\n",
    "        ) AS \"final_edit_sum\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE channel = '#de.wikipedia'\n",
    "GROUP BY 1, 2\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a5fe00ae-aacd-409e-baba-e8fb2ca5586b",
   "metadata": {},
   "source": [
    "Combining what you have seen on cumulative calculations, try adapting the query above to incorporate a new column that begins with zero edits, and then reaches `final_edit_sum`. Can you make it work on a per-channel for all channels that begin with `#e`?"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1ff2fb13-76c3-4137-93e6-93fd84fdd32d",
   "metadata": {},
   "source": [
    "## Re-using window definitions\n",
    "\n",
    "Use WINDOW to create window definitions that you can re-use in your SQL statement.\n",
    "\n",
    "The query above has been reworked to abstract the common window definitions:\n",
    "\n",
    "* `channelTime_window` is partitioned by channel and ordered by time.\n",
    "* `channel_window` is partitioned by channel.\n",
    "* The LAG, LEAD, FIRST_VALUE, and LAST_VALUE functions have been adapted to address the WINDOW definitions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3f6586ac-7fe5-4f94-9672-b458f33d3ce9",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = f\"\"\"\n",
    "SELECT TIME_FLOOR(\"__time\",'PT1H') AS \"period\",\n",
    "    channel,\n",
    "    SUM(delta) AS \"edit_sum\",\n",
    "    LAG(SUM(delta)) OVER channelTime_window AS \"previous_edit_sum\",\n",
    "    LEAD(SUM(delta), 2) OVER channelTime_window AS \"nextNext_edit_sum\",\n",
    "    FIRST_VALUE(SUM(delta)) OVER channel_window AS \"first_edit_sum\",\n",
    "    LAST_VALUE(SUM(delta)) OVER channel_window AS \"final_edit_sum\"\n",
    "FROM \"{table_name}\"\n",
    "WHERE channel = '#de.wikipedia'\n",
    "GROUP BY 1, 2\n",
    "WINDOW\n",
    "    channelTime_window AS\n",
    "    (\n",
    "        PARTITION BY channel\n",
    "        ORDER BY TIME_FLOOR(\"__time\",'PT1H')\n",
    "    ),\n",
    "    channel_window AS\n",
    "    (\n",
    "        PARTITION BY channel\n",
    "    )\n",
    "\"\"\"\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a10abe1c-ec46-4010-8a36-65cc385f753c",
   "metadata": {},
   "source": [
    "Take a moment to review some of the queries in this notebook. Can you simplify them by using WINDOW?"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fab5b831-69a8-4215-9b97-f7ce62610850",
   "metadata": {},
   "source": [
    "## Clean up\n",
    "\n",
    "Run the following cell to remove the XXX used in this notebook from the database."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8947babc-0798-465d-8eed-b9facca064d0",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Drop table: [{druid.datasources.drop(table_name)}]\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc5212e9-35ed-4243-bc98-7f65afcfaf22",
   "metadata": {},
   "source": [
    "## Summary\n",
    "\n",
    "* Windows are defined in an OVER clause.\n",
    "* The data in windows can be ordered and / or partitioned.\n",
    "* Many standard aggregations can operate over windows.\n",
    "* Specific window functions exist for ranking, numbering, and bucketing results.\n",
    "* There are additional functions that return values from specific rows in the window.\n",
    "* Windows can be defined inline or by using WINDOW.\n",
    "\n",
    "## Learn more\n",
    "\n",
    "* Amend the queries in this notebook to try other functions, such as PERCENT_RANK.\n",
    "* Try out different definitions for windows from the [official documentation](https://druid.apache.org/docs/latest/querying/sql-window-functions/#window-function-syntax), such as setting a scope to UNBOUNDED FOLLOWING."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.6"
  },
  "vscode": {
   "interpreter": {
    "hash": "b0fa6594d8f4cbf19f97940f81e996739fb7646882a419484c72d19e05852a7e"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}