notebooks/03-query/11-joins.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8a6c7193-0499-4f89-b6c8-b0e6a1634149",
   "metadata": {},
   "source": [
    "# Using joins effectively in Druid \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",
    "Apache Druid can do joins at query time through both its native engine (scatter/gather) and through the Multi-Stage Query engine.\n",
    "Note that the best performance will always be without using any joins, so the overall recommendation is to join the data at ingestion when loading in batch, or upstream of ingestion for streaming scenarios. But pre-joining the data isn't always an option, so you'll need to learn how to use Druid to process joins as efficiently as possible for those cases.\n",
    "\n",
    "Through this notebook you will run tests on the performance of different approaches to joining data, seeing how different JOIN operations are processed and how to write SQL that will take advantage of how it works in each case.\n",
    "The notebook also covers alternatives to joins using UNION ALL strategies or application side code. \n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bbdbf6ad-ca7b-40f5-8ca3-1070f4a3ee42",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "## Prerequisites\n",
    "\n",
    "This tutorial works with Druid 29.0.0 or later.\n",
    "\n",
    "#### Run with Docker\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 project page](https://github.com/implydata/learn-druid).\n",
    "   "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5007a243-b81a-4601-8f57-5b14940abbff",
   "metadata": {},
   "source": [
    "## Initialization\n",
    "\n",
    "The following cells set up the notebook and learning environment ready for use.\n",
    "\n",
    "### Set up and connect to the learning environment\n",
    "\n",
    "Run the next cell to set up the Druid Python client's connection to Apache Druid and a client to interact with the Data Generator.\n",
    "\n",
    "If successful, the Druid version number will be shown in the output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c1ec783b-df3f-4168-9be2-cdc6ad3e33c2",
   "metadata": {},
   "outputs": [],
   "source": [
    "import druidapi\n",
    "import os\n",
    "import json\n",
    "import pandas\n",
    "import time\n",
    "from datetime import datetime, timedelta\n",
    "import requests\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",
    "\n",
    "display = druid.display\n",
    "sql_client = druid.sql\n",
    "status_client = druid.status\n",
    "\n",
    "print(f\"Version:{status_client.version}\")\n",
    "\n",
    "# client for Data Generator API\n",
    "datagen = druidapi.rest.DruidRestClient(\"http://datagen:9999\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "96394476-0834-4396-9e89-c7dd8c268b3f",
   "metadata": {},
   "source": [
    "The following helper function is used throughout the examples in this notebook to provide a timeline of execution that can then be compared to the Docker container stats history to examine CPU, Memory, I/O and Network activity for each operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5a8d44cc-91c9-42c0-872d-4f5c426a0456",
   "metadata": {},
   "outputs": [],
   "source": [
    "#function to record progress timestamps so they can be compared to resource utilization metrics\n",
    "def print_timestamp( msg = '' ):\n",
    "    t = datetime.now()\n",
    "    print(f\"{t} <<<<<<<<<<<<< {msg}\")\n",
    "    return t"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e06f1b86-83b4-4248-9ac7-d47ad155cbca",
   "metadata": {},
   "source": [
    "## Generate data\n",
    "Run the following cells which will:\n",
    "- Data for clicks using the data generators `clickstream/clickstream.json` configuration.\n",
    "- Data for users with the `clickstream/users_init.json` configuration.\n",
    "These two data generator configs are compatible because they share a domain for the user_id field values. In other words, they are joinable.\n",
    "  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f968118a-36cc-4eba-94c0-428194dee59a",
   "metadata": {},
   "outputs": [],
   "source": [
    "def wait_for_datagen( job_name:str):\n",
    "    import time\n",
    "    from IPython.display import clear_output\n",
    "    # wait for the messages to be fully published \n",
    "    done = False\n",
    "    while not done:\n",
    "        result = datagen.get_json(f\"/status/{job_name}\",'')\n",
    "        clear_output(wait=True)\n",
    "        print(json.dumps(result, indent=2))\n",
    "        if result[\"status\"] == 'COMPLETE':\n",
    "            done = True\n",
    "        else:\n",
    "            time.sleep(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "088d2695-cce8-4f52-a601-90953356d5ce",
   "metadata": {},
   "outputs": [],
   "source": [
    "print_timestamp()\n",
    "\n",
    "# simulate clicks for last 2 days\n",
    "gen_hours=48\n",
    "gen_now = datetime.now() - timedelta(hours=gen_hours)\n",
    "gen_start_time = gen_now.strftime(\"%Y-%m-%d %H:%M:%S\")\n",
    "\n",
    "headers = {\n",
    "  'Content-Type': 'application/json'\n",
    "}\n",
    "\n",
    "datagen_request = {\n",
    "    \"name\": \"clicks\",\n",
    "    \"target\": { \"type\": \"file\", \"path\":\"clicks.json\"  },\n",
    "    \"config_file\": \"clickstream/clickstream.json\", \n",
    "    \"time\": f\"{gen_hours}h\",\n",
    "    \"concurrency\":200,\n",
    "    \"time_type\":gen_start_time\n",
    "}\n",
    "datagen.post(\"/start\", json.dumps(datagen_request), headers=headers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "25915faa-1953-44ab-9898-5bc3798ebdde",
   "metadata": {},
   "outputs": [],
   "source": [
    "wait_for_datagen( \"clicks\")  # this takes about 2 minutes\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "adbbe00c-2b79-47d4-91df-1d93f942b79f",
   "metadata": {},
   "outputs": [],
   "source": [
    "print_timestamp()\n",
    "datagen_request = {\n",
    "    \"name\": \"users\",\n",
    "    \"target\": { \"type\": \"file\", \"path\":\"users.json\"  },\n",
    "    \"config_file\": \"clickstream/users_init.json\", \n",
    "    \"concurrency\":100,\n",
    "    \"total_events\":4000 \n",
    "}\n",
    "datagen.post(\"/start\", json.dumps(datagen_request), headers=headers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2beda33c-6f22-456d-9189-681d547cca8a",
   "metadata": {},
   "outputs": [],
   "source": [
    "wait_for_datagen( \"users\")\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ddd8faf9-a267-4607-83a0-85c7720c39cc",
   "metadata": {},
   "source": [
    "The data generation `target`s above use files \"clicks.json\" and \"users.json\" respectively as output files. These files are generated locally on the data generation server and can be accessed through HTTP at: \n",
    "- http://localhost:9999/file/clicks.json\n",
    "- http://localhost:9999/file/users.json \n",
    "\n",
    "You will use those URLs below in the ingestion SQL's EXTERN table functions to load this data.\n",
    "\n",
    "### Load `clicks` and `users` into separate tables\n",
    "\n",
    "Load separate tables for `clicks` and `users` to demonstrate join functionality."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6beb309b-2a1a-43c7-a001-86b61e47adf4",
   "metadata": {},
   "source": [
    "### Execute the batch ingestions\n",
    "This cell uses the files you generated above to load the files into respective tables `clicks` and `users`.\n",
    "The files are referenced in the `EXTERN` table function using an [\"http\" typed input source](https://druid.apache.org/docs/latest/ingestion/input-sources#http-input-source)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2f1d9978-ff4f-4b77-b926-b42ce17419a3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# ingest users\n",
    "sql='''\n",
    "REPLACE INTO \"users\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"http://datagen:9999/file/users.json\"]}',\n",
    "    '{\"type\":\"json\"}'\n",
    "  )\n",
    ") EXTEND (\"time\" VARCHAR, \"user_id\" VARCHAR, \"first_name\" VARCHAR, \"last_name\" VARCHAR, \"dob\" VARCHAR, \"address_lat\" VARCHAR, \"address_long\" VARCHAR, \"marital_status\" VARCHAR, \"income\" VARCHAR, \"signup_ts\" VARCHAR))\n",
    "SELECT\n",
    "  TIMESTAMP'1970-01-01 00:00:00' AS \"__time\",\n",
    "  \"user_id\",\n",
    "  \"first_name\",\n",
    "  \"last_name\",\n",
    "  \"dob\",\n",
    "  \"address_lat\",\n",
    "  \"address_long\",\n",
    "  \"marital_status\",\n",
    "  \"income\",\n",
    "  \"signup_ts\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY ALL\n",
    "'''\n",
    "display.run_task(sql)\n",
    "print_timestamp()\n",
    "\n",
    "# ingest clicks\n",
    "sql='''\n",
    "REPLACE INTO \"clicks\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"http://datagen:9999/file/clicks.json\"]}',\n",
    "    '{\"type\":\"json\"}'\n",
    "  )\n",
    ") EXTEND (\"time\" VARCHAR, \"user_id\" VARCHAR, \"event_type\" VARCHAR, \"client_ip\" VARCHAR, \"client_device\" VARCHAR, \"client_lang\" VARCHAR, \"client_country\" VARCHAR, \"referrer\" VARCHAR, \"keyword\" VARCHAR, \"product\" VARCHAR))\n",
    "SELECT\n",
    "  TIME_PARSE(\"time\") AS \"__time\",\n",
    "  \"user_id\",\n",
    "  \"event_type\",\n",
    "  \"client_ip\",\n",
    "  \"client_device\",\n",
    "  \"client_lang\",\n",
    "  \"client_country\",\n",
    "  \"referrer\",\n",
    "  \"keyword\",\n",
    "  \"product\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY DAY\n",
    "'''\n",
    "display.run_task(sql)\n",
    "print_timestamp()\n",
    "\n",
    "# make sure both tables are available before moving on\n",
    "sql_client.wait_until_ready('users')\n",
    "sql_client.wait_until_ready('clicks')\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "472589e4-1026-4b3b-bb79-eedabb2b44c4",
   "metadata": {},
   "source": [
    "### Pre-join into denormalized table\n",
    "In many cases it is best to pre-join the data. Druid is particularly suited for analytic applications that require subsecond response times in order to drive an interactive user experience with high concurrency. In a parallel system, avoiding query time joins is one of the tools that is useful to achieve high query performance at high concurrencies by avoiding the movement of data during queries.\n",
    "\n",
    "The next cell ingests the two datasets into a single pre-joined table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d78f9079-0816-4454-a3a1-267a60be8f75",
   "metadata": {},
   "outputs": [],
   "source": [
    "# ingest clicks_enhanced including user data\n",
    "sql='''\n",
    "REPLACE INTO \"clicks_enhanced\" OVERWRITE ALL\n",
    "WITH \n",
    "\"users_ext\" AS \n",
    "(\n",
    "  SELECT *\n",
    "  FROM TABLE(\n",
    "    EXTERN(\n",
    "      '{\"type\":\"http\",\"uris\":[\"http://datagen:9999/file/users.json\"]}',\n",
    "      '{\"type\":\"json\"}'\n",
    "    )\n",
    "  ) EXTEND (\"time\" VARCHAR, \"user_id\" VARCHAR, \"first_name\" VARCHAR, \"last_name\" VARCHAR, \"dob\" VARCHAR, \"address_lat\" VARCHAR, \"address_long\" VARCHAR, \"marital_status\" VARCHAR, \"income\" VARCHAR, \"signup_ts\" VARCHAR)\n",
    "),\n",
    "\"clicks_ext\" AS \n",
    "(\n",
    "  SELECT *\n",
    "  FROM TABLE(\n",
    "    EXTERN(\n",
    "      '{\"type\":\"http\",\"uris\":[\"http://datagen:9999/file/clicks.json\"]}',\n",
    "      '{\"type\":\"json\"}'\n",
    "    )\n",
    "  ) EXTEND (\"time\" VARCHAR, \"user_id\" VARCHAR, \"event_type\" VARCHAR, \"client_ip\" VARCHAR, \"client_device\" VARCHAR, \"client_lang\" VARCHAR, \"client_country\" VARCHAR, \"referrer\" VARCHAR, \"keyword\" VARCHAR, \"product\" VARCHAR)\n",
    ")\n",
    "SELECT\n",
    "  TIME_PARSE(c.\"time\") AS \"__time\",\n",
    "  c.\"user_id\",\n",
    "  c.\"event_type\",\n",
    "  c.\"client_ip\",\n",
    "  c.\"client_device\",\n",
    "  c.\"client_lang\",\n",
    "  c.\"client_country\",\n",
    "  c.\"referrer\",\n",
    "  c.\"keyword\",\n",
    "  c.\"product\",\n",
    "  u.\"first_name\",\n",
    "  u.\"last_name\",\n",
    "  u.\"dob\",\n",
    "  TIMESTAMPDIFF(YEAR, TIME_PARSE(u.\"dob\"), CURRENT_TIMESTAMP) as age,\n",
    "  ROUND( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.\"dob\"), CURRENT_TIMESTAMP), -1) as age_group,\n",
    "  u.\"address_lat\",\n",
    "  u.\"address_long\",\n",
    "  u.\"marital_status\",\n",
    "  u.\"income\",\n",
    "  u.\"signup_ts\"\n",
    "FROM \"clicks_ext\" c LEFT OUTER JOIN \"users_ext\" u ON c.\"user_id\"=u.\"user_id\"\n",
    "PARTITIONED BY ALL\n",
    "'''\n",
    "display.run_task(sql)\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "350c4465-1196-4fa4-98d6-5a245d4afc20",
   "metadata": {},
   "source": [
    "### Load users as a lookup\n",
    "Lookups are in-memory key/value tables that are pre-broadcast to all brokers, peons and historicals that belong to the lookup tier. They are faster to join to other tables because they are already in memory where they are needed.\n",
    "The following cells prepare and load the user_age_group lookup to demonstrate the benefits of this join strategy further below in this notebook.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f52a94fb-d2e4-403f-ab10-84d3af7bf2c8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define helper functions to load lookup\n",
    "def postLookup(definition):\n",
    "    x = requests.post(druid_host + '/druid/coordinator/v1/lookups/config', json=definition)\n",
    "\n",
    "    if \"error\" in x.text:\n",
    "        raise Exception('Not able to complete the request. \\n\\n'+x.text)\n",
    "    else:\n",
    "        print('Successfully submitted the lookup request.')\n",
    "\n",
    "def waitForLookup(tier, name, ticsMax):\n",
    "\n",
    "    # The default time period between checks of lookup definition changes (druid.manager.lookups.period)\n",
    "    # is two minutes. The notebook environment reduces this for learning purposes.\n",
    "    # \n",
    "    # https://druid.apache.org/docs/27.0.0/configuration/#lookups-dynamic-configuration\n",
    "\n",
    "    tics = 0\n",
    "    ticsWait = 1    \n",
    "    ticsMax = min(ticsMax,360)\n",
    "    ticsSpinner = \"/-\\|\"\n",
    "    \n",
    "    apicall = druid_host + '/druid/coordinator/v1/lookups/status/'+tier+'/'+name+'?detailed=true'\n",
    "\n",
    "    x = requests.get(apicall)\n",
    "\n",
    "    while (x.text != '{\"loaded\":true,\"pendingNodes\":[]}' and tics < ticsMax):\n",
    "        print(x.text + ' ' + ticsSpinner[tics%len(ticsSpinner)] + ' [' + str(ticsMax-tics) + ']   ', end='\\r')\n",
    "        time.sleep(ticsWait)\n",
    "        tics += 1\n",
    "        x = requests.get(apicall) \n",
    "\n",
    "    if (tics == ticsMax):\n",
    "        raise Exception('\\nTimeout waiting for Druid to load the ' + name + ' lookup to ' + tier + 'tier. Run the cell again.')\n",
    "    else:\n",
    "        print('\\nSuccess. ' + name + ' lookup in ' + tier + ' tier is fully available.')\n",
    "        \n",
    "# initialize lookup environment        \n",
    "empty_post = {}\n",
    "postLookup(empty_post)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7e115d67-d6db-4163-9224-0a8963337715",
   "metadata": {},
   "outputs": [],
   "source": [
    "# define user_age_group lookup by building it from a query of users\n",
    "sql='''\n",
    "SELECT user_id, ROUND( TIMESTAMPDIFF(YEAR, TIME_PARSE(\"dob\"), CURRENT_TIMESTAMP),-1) as age_group\n",
    "FROM \"users\"\n",
    "'''\n",
    "results = sql_client.sql(sql)\n",
    "map = dict( zip([ r['user_id'] for r in results ], [ r['age_group'] for r in results ]))\n",
    "\n",
    "lookup_tier = \"__default\"\n",
    "lookup_name = \"user_age_group\"\n",
    "lookup_definition_version = datetime.now().strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n",
    "lookup_definition = {\n",
    "    lookup_tier: {\n",
    "        lookup_name: {\n",
    "            \"version\": lookup_definition_version,  \n",
    "            \"lookupExtractorFactory\": {\n",
    "                \"type\": \"map\",\n",
    "                    \"map\": map\n",
    "            }\n",
    "        }\n",
    "    }\n",
    "}\n",
    "postLookup(lookup_definition)\n",
    "waitForLookup(lookup_tier, lookup_name, 30)\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6bf7f21b-7cdd-460e-bc00-e74037e21717",
   "metadata": {},
   "source": [
    "## Prep query testing and measurement function\n",
    "We measure each method of joining the data by running the query multiple times and avoiding caching of the results in order to get a avg, min and max runtimes of each query. We'll use the following function to achieve this:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "38f8b68a-52ca-41bf-9d84-ea668632528c",
   "metadata": {},
   "outputs": [],
   "source": [
    "from statistics import mean \n",
    "\n",
    "def measure_query( sql: str, iterations: int ):\n",
    "    req = sql_client.sql_request(sql)\n",
    "    req.add_context(\"populateCache\", \"false\")  # run without caching results to get a real sense of performance\n",
    "    req.add_context(\"useCache\", \"false\")  # do not use cached results\n",
    "    stats = []\n",
    "    while (iterations>0):\n",
    "      start = datetime.now()\n",
    "      sql_client.sql(req)\n",
    "      end = datetime.now()\n",
    "      stats.append( (end - start).total_seconds() * 10**3 ) # add run time in milliseconds\n",
    "      iterations -=1\n",
    "    return f\"Results = avg:{mean(stats)} ms   min:{min(stats)} ms  max:{max(stats)} ms\""
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "4cf7592d-b852-4712-b8ec-60be9df1224d",
   "metadata": {},
   "source": [
    "## Synchronous queries on scatter/gather engine\n",
    "Druid's scatter/gather query processing is designed for speed.\n",
    "### The scatter/gather engine\n",
    "At a high level, Druid queries run as follows:\n",
    "\n",
    "<img src=\"assets/scatter-gather.png\" alt=\"drawing\" style=\"width:600px;\"/>\n",
    "\n",
    "1. Broker receives query\n",
    "2. Broker plans query and determines which segments are relevant using cached segment metadata\n",
    "3. Broker forwards query to data servers, telling each server what segments to interrogate\n",
    "4. Data servers process segment files\n",
    "5. Data servers send results back to Broker\n",
    "6. Broker performs final processing and returns results\n",
    "\n",
    "Note that the data servers do not directly communicate with each other.\n",
    "\n",
    "### Joins with a scatter/gather engine\n",
    "When processing a join, the engine uses multiple scatter/gather steps. It broadcasts the results from the first scatter/gather in a second scatter/gather. The first table in the FROM clause is processed last which means its data does not need to move in order to process the join. This means that you should put the largest table at the front of the joins.\n",
    "![](assets/join-scatter-gather.png)\n",
    "1. The Broker receives the query request and creates a plan. The first datasource in the from clause becomes the driving table, meaning that it will process the join on this table in parallel across all Data Servers involved.\n",
    "3. The Broker reads `users` from all data servers and merges them into a single dataset.\n",
    "4. Merged `users` are broadcast to all data servers involved by adding them as an inline data source to the driving table request that is submitted to Data Servers.\n",
    "5. Data Servers join the `clicks` data they hold with the inlined `users` which have all users.\n",
    "6. To finish up this query, each Data Server processes its partial aggregation.\n",
    "7. Finally, the Broker does final aggregation and returns result to the user.\n",
    "\n",
    "Try it out:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5d0a4a59-ffbd-4ab2-944d-31275ca5ef24",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, \n",
    "  round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) as age_group, \n",
    "  count(distinct c.client_ip) as distint_ips\n",
    "FROM clicks c \n",
    "     INNER JOIN users u\n",
    "        ON c.user_id=u.user_id\n",
    "WHERE c.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "GROUP BY 1, round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) \n",
    "ORDER BY 3 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "display.sql(sql)\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "93cb2f03-0831-41ae-8df8-865a77280aa2",
   "metadata": {},
   "source": [
    "Run it 20 times..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7b97cdfe-f46b-4cd9-8b59-faba4a4c015e",
   "metadata": {},
   "outputs": [],
   "source": [
    "join_results=[]\n",
    "result = measure_query(sql, 20)\n",
    "# saving all results for comparison \n",
    "join_results.append({\"query_type\":\"simple_join\",\"results\":result})\n",
    "print(result)\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a040384d-3d46-4bce-a084-c020433fa676",
   "metadata": {},
   "source": [
    "### Order of the joins is important\n",
    "\n",
    "![](assets/fail-scatter-gather.png)\n",
    "\n",
    "If you change the order of the joins, it will fail because the subquery will return more than `maxSubQueryRows` which defaults to 100000. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d8184fcf-a2cb-44b6-af0d-6bd9511fbd11",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, \n",
    "  round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) as age_group, \n",
    "  count(distinct c.client_ip) as distint_ips\n",
    "FROM users u\n",
    "     INNER JOIN clicks c\n",
    "        ON c.user_id=u.user_id\n",
    "WHERE c.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "GROUP BY 1, round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) \n",
    "ORDER BY 3 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "try:\n",
    "  display.sql(sql)\n",
    "except Exception as ex:\n",
    "  print(f\"ERROR: {ex}\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "92cb0f7e-60a1-4a7c-9060-08748cf35036",
   "metadata": {},
   "source": [
    "### Nested subqueries\n",
    "See the query in the next cell, Druid will execute the subquery first as set of scatter/gather steps and then process the rest of the joins in subsequent scatter/gather steps. The results of the subquery still needs to be within the `maxSubQueryRows` limit, but running it this way will allow all processing of the larger `clicks` data with the parallelism available on the Data Servers: \n",
    "![](assets/joins-subquery.png)\n",
    "\n",
    "The subquery calculates the total number of clicks per age group by day for for that last day. The outer query then calculates this by user and calculates each user's contribution percentage of the overall activity. The result shows the top 10 users by contribution percentage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bbff92a3-c903-43b8-9536-b08dacf9941b",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, c.user_id,\n",
    "  round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) as age_group, \n",
    "  count(1) as user_clicks,\n",
    "  count(1)*100.0/agg.clicks as age_group_contrib_pct\n",
    "FROM clicks c\n",
    "     INNER JOIN users u ON c.user_id = u.user_id\n",
    "     INNER JOIN\n",
    "     (\n",
    "         SELECT TIME_FLOOR(cc.__time, 'P1D') as _date,\n",
    "                ROUND( TIMESTAMPDIFF(YEAR, TIME_PARSE(uu.dob), CURRENT_TIMESTAMP),-1) as age_group,\n",
    "                COUNT(1) as clicks\n",
    "           FROM clicks cc\n",
    "           INNER JOIN users uu ON cc.user_id=uu.user_id\n",
    "           WHERE cc.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "           GROUP BY 1, ROUND( TIMESTAMPDIFF(YEAR, TIME_PARSE(uu.dob), CURRENT_TIMESTAMP),-1)\n",
    "     ) agg\n",
    "        ON agg._date = TIME_FLOOR(c.__time, 'P1D') AND agg.age_group=round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) \n",
    "WHERE c.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "GROUP BY 1, 2, round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1), agg.clicks \n",
    "ORDER BY 1, 5 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "display.sql(sql)\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a96c99be-75c6-410d-9143-0588d887d6ff",
   "metadata": {},
   "outputs": [],
   "source": [
    "result = measure_query(sql, 20)\n",
    "# saving all results for comparison \n",
    "join_results.append({\"query_type\":\"subquery_join\",\"results\":result})\n",
    "print_timestamp()\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1ee64c9d-4f6d-42b9-827f-4ac01076e00e",
   "metadata": {},
   "source": [
    "### Join using a lookup\n",
    "Lookups in Druid are broadcast to all data servers at the time of creation and can be refreshed [continuously from a kafka stream](https://druid.apache.org/docs/latest/development/extensions-core/kafka-extraction-namespace) or refreshed periodically. The execution of the query can now be done in a single scatter/gather step:\n",
    "![](assets/joins-lookup.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "19c38a96-d237-4aab-b33d-deac42f80758",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, \n",
    "  u.v as age_group, \n",
    "  count(distinct c.client_ip) as distint_ips\n",
    "FROM clicks c \n",
    "     INNER JOIN lookup.user_age_group u\n",
    "        ON c.user_id=u.k\n",
    "WHERE c.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "GROUP BY 1, 2\n",
    "ORDER BY 3 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "\n",
    "result = measure_query(sql, 20)\n",
    "# saving all results for comparison \n",
    "join_results.append({\"query_type\":\"simple_join_lookup\",\"results\":result})\n",
    "print_timestamp()\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dc877650-7728-44c7-9aa9-5f67ac606503",
   "metadata": {},
   "source": [
    "... and the second query using the `lookup`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0aee9a0e-fa53-4166-a756-8b24c13ccb2d",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = '''\n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, c.user_id,\n",
    "  u.v as age_group, \n",
    "  count(1) as user_clicks,\n",
    "  count(1)*100.0/agg.clicks as age_group_contrib_pct\n",
    "FROM clicks c\n",
    "     INNER JOIN lookup.user_age_group u ON c.user_id = u.k\n",
    "     INNER JOIN\n",
    "     (\n",
    "         SELECT TIME_FLOOR(cc.__time, 'P1D') as _date,\n",
    "                uu.v as age_group,\n",
    "                COUNT(1) as clicks\n",
    "           FROM clicks cc\n",
    "           INNER JOIN lookup.user_age_group uu ON cc.user_id=uu.k\n",
    "           GROUP BY 1, 2\n",
    "     ) agg\n",
    "        ON agg._date = TIME_FLOOR(c.__time, 'P1D') AND agg.age_group=u.v \n",
    "WHERE c.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "GROUP BY 1, 2, 3, agg.clicks \n",
    "ORDER BY 1, 5 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "result = measure_query(sql, 20)\n",
    "# saving all results for comparison \n",
    "join_results.append({\"query_type\":\"subquery_join_lookup\",\"results\":result})\n",
    "print_timestamp()\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "62205ce8-a1c4-4256-81b3-f6c287e4017a",
   "metadata": {},
   "source": [
    "### Query pre-joined data\n",
    "Avoiding the join altogether is usually the best option.\n",
    "In the simply query case there are no joins because we processed the join and calculated the user's age_group at the time of ingestion. This saves a lot of processing at query time. You should always consider this option with Druid because while it will increase the size of the data (see below for different storage footprints of each method), it will save a lot of processing CPU and memory at query time.\n",
    "Here are the same 2 queries but re-written using the `clicks_enhanced` table which we joined at ingestion above:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "11b37e2b-68be-43a3-9a53-7734ce27238a",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql='''\n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, \n",
    "  c.age_group, \n",
    "  count(distinct c.client_ip) as distint_ips\n",
    "FROM clicks_enhanced c \n",
    "WHERE c.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "GROUP BY 1, 2\n",
    "ORDER BY 3 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "\n",
    "result = measure_query(sql, 20)\n",
    "# saving all results for comparison \n",
    "join_results.append({\"query_type\":\"simple_prejoin\",\"results\":result})\n",
    "print_timestamp()\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d013699b-c3b9-497d-9712-fdb536a93206",
   "metadata": {},
   "source": [
    "...even with the subquery to calculate totals by age group by day, it still avoids two joins by using the pre-joined datasource:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "97ef67de-20a9-4aa0-b835-a0142cd8a0bd",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = '''\n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, c.user_id,\n",
    "  c.age_group, \n",
    "  count(1) as user_clicks,\n",
    "  count(1)*100.0/agg.clicks as age_group_contrib_pct\n",
    "FROM clicks_enhanced c\n",
    "     INNER JOIN\n",
    "     (\n",
    "         SELECT TIME_FLOOR(cc.__time, 'P1D') as _date,\n",
    "                cc.age_group,\n",
    "                COUNT(1) as clicks\n",
    "           FROM clicks_enhanced cc\n",
    "           GROUP BY 1, 2\n",
    "     ) agg\n",
    "        ON agg._date = TIME_FLOOR(c.__time, 'P1D') AND agg.age_group=c.age_group \n",
    "WHERE c.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "GROUP BY 1, 2, 3, agg.clicks \n",
    "ORDER BY 1, 5 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "result = measure_query(sql, 20)\n",
    "# saving all results for comparison \n",
    "join_results.append({\"query_type\":\"subquery_prejoin\",\"results\":result})\n",
    "print_timestamp()\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "64372aea-9915-4e8a-a4ec-b34b160bd3b3",
   "metadata": {},
   "source": [
    "### Results by join strategy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "81807f87-cff5-4e1c-8dd7-cc26b2b169e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# review the results\n",
    "sorted_list = sorted(join_results, key=lambda x: x['query_type'])\n",
    "sorted_list"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "66fd0a23-cc14-4537-8965-5dab8eeb1b46",
   "metadata": {},
   "source": [
    "In these results you can see the progression of response times. \n",
    "- The use of lookups improves performance over joining to another datasource, but it requires enough Heap to fit all lookups on all historical and ingestion tasks (Peon) JVMs.\n",
    "- The pre-joined table performs even better and has the added advantage of not requiring any additional Heap.\n",
    "\n",
    "If the data being joined changes over time and you wish to query with the latest values, consider a lookup. If the data does not change or the change is not desirable, then pre-join the data. The example of `age_group` as the joined property works well in the pre-join strategy because the user's age_group is important at the time they clicked, not when looking at old click rows based on their current age_group."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "92e6116f-3632-4b1f-817f-d656f598758e",
   "metadata": {},
   "source": [
    "<a id='join_with_inequality'></a>\n",
    "### INNER JOIN with arbitrary conditions\n",
    "\n",
    "Before the 29.0.0 release, all join conditions had to be equalities. With the new functionality inequalities can also be used. The query optimization rules continue to apply, so the equality portion of the join will be processed as described above. The inequalities are processed after the join operation is completed.\n",
    "\n",
    "In this first example, the query finds activity within a limited the geographical area by filtering for a range of latitudes where the user's is located:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f29569b0-b9a3-4672-866a-af75ae47edc9",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = ''' \n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, \n",
    "  round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) as age_group, \n",
    "  count(distinct c.client_ip) as distinct_ips\n",
    "FROM clicks c\n",
    "     INNER JOIN users u\n",
    "        ON c.user_id=u.user_id AND u.address_lat > -1 AND u.address_lat < 1\n",
    "GROUP BY 1, round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) \n",
    "ORDER BY 3 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "\n",
    "display.sql(sql)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1352b491-1795-44dd-91a3-ff75de61f24f",
   "metadata": {},
   "source": [
    "This second example looks at clicks from the same user during the same hour but for a different country, trying to find users with potentially shared accounts or breached accounts. \n",
    "This requires that we have a more complex join condition when self joining the clicks data: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6d77125a-2b77-4bb9-9ec4-684ce29f8e2e",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = ''' \n",
    "WITH clicks_by_country AS\n",
    "( \n",
    "SELECT TIME_FLOOR(__time, 'PT1H') \"hour\", \"user_id\", \"client_country\", count(*) as \"click_count\"\n",
    " FROM clicks_enhanced \n",
    " WHERE __time > CURRENT_TIMESTAMP - INTERVAL '1' DAY \n",
    " GROUP BY 1,2,3\n",
    ")\n",
    "SELECT c1.\"hour\", c1.\"user_id\", \n",
    "       c1.\"client_country\" as \"first_country\", \n",
    "       c2.\"client_country\" as \"second_country\",\n",
    "       c1.\"click_count\" as \"first_country_clicks\",\n",
    "       c2.\"click_count\" as \"second_country_clicks\"\n",
    "FROM \n",
    "  clicks_by_country c1\n",
    "INNER JOIN\n",
    "  clicks_by_country c2\n",
    "  ON c1.\"hour\" = c2.\"hour\" AND c1.\"user_id\"=c2.\"user_id\" \n",
    "    AND c1.\"client_country\" <> c2.\"client_country\"\n",
    "    AND c1.\"click_count\" > c2.\"click_count\"\n",
    "LIMIT 10\n",
    "'''\n",
    "\n",
    "display.sql(sql)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46e2e6f6-86e1-48dd-a125-6c1920173d74",
   "metadata": {},
   "source": [
    "## Asynchronous queries on the MSQ engine\n",
    "\n",
    "The MSQ (Multi-stage query) engine works by shuffling rows between stages. All stages execute in tasks in parallel as much as they can.\n",
    "![](assets/msq-engine.png)\n",
    "\n",
    "Typically, as you did in the ingestion above, the MSQ engine is used to ingest external data, but it can also be used to query Druid datasources directly. It has the advantage of being able to do either `broadcast` or `sortMerge` joins. \n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e8b27edf-90f2-4f69-90e1-c5db281f2934",
   "metadata": {},
   "source": [
    "### Broadcast joins in MSQ\n",
    "Broadcast joins execute similarly to the gather/scatter approach in that they send the whole right-hand dataset to all workers, but MSQ runs asynchronously and does not have the `maxSubQueryRows` limit. It will broadcast and load the whole right hand side table on all query_worker tasks.\n",
    "![](assets/msq-broadcast.png)\n",
    "\n",
    "In the first stage, `users` segments are scanned and all rows are broadcast to all workers. `clicks` data does not need to be distributed because all `users` rows are now available in all workers so join processing can proceed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "711ffa98-063d-4c10-ade4-65f6d609b1ad",
   "metadata": {},
   "outputs": [],
   "source": [
    "def measure_async_query( sql: str, iterations: int, joinAlgorithm = 'broadcast', workers=3 ):\n",
    "    req = sql_client.sql_request(sql)\n",
    "    req.add_context(\"populateCache\", \"false\")  # run without caching results to get a real sense of performance\n",
    "    req.add_context(\"useCache\", \"false\")  # do not use cached results\n",
    "    req.add_context(\"sqlJoinAlgorithm\", joinAlgorithm)\n",
    "    req.add_context(\"maxNumTasks\", workers)\n",
    "    stats = []\n",
    "    while (iterations>0):\n",
    "      start = datetime.now()\n",
    "      res=sql_client.async_sql(req, rowsPerPage=100)\n",
    "      res.wait_until_done()\n",
    "      end = datetime.now()\n",
    "      stats.append( (end - start).total_seconds() * 10**3 ) # add run time in milliseconds\n",
    "      iterations -=1\n",
    "    return f\"Results = avg:{mean(stats)} ms   min:{min(stats)} ms  max:{max(stats)} ms\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a36e50b0-ba36-4df2-bdf5-bea30424c420",
   "metadata": {},
   "outputs": [],
   "source": [
    "#MSQ async query using broadcast join\n",
    "sql='''\n",
    "SELECT TIME_FLOOR(c.__time, 'P1D') as _date, \n",
    "  round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) as age_group, \n",
    "  count(distinct c.client_ip) as distint_ips\n",
    "FROM clicks c\n",
    "     INNER JOIN users u\n",
    "        ON c.user_id=u.user_id\n",
    "WHERE c.__time > CURRENT_TIMESTAMP  - INTERVAL '1' DAY\n",
    "GROUP BY 1, round( TIMESTAMPDIFF(YEAR, TIME_PARSE(u.dob), CURRENT_TIMESTAMP),-1) \n",
    "ORDER BY 3 DESC\n",
    "LIMIT 10\n",
    "'''\n",
    "res = sql_client.async_sql(sql)\n",
    "res.wait_until_done()\n",
    "result = pandas.json_normalize(res.rows)\n",
    "print_timestamp()\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2db4fe7e-47de-40eb-b3fa-2af7846a1638",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(measure_async_query(sql,10)) # using default broadcast join\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bdf9ad82-4098-4fab-bc7d-3b91068ee73e",
   "metadata": {},
   "source": [
    "MSQ has an inherent disadvantage over Scatter/Gather when it comes to response time for many queries. It must first initialize the `query_controller` and `query_worker` tasks and only then begin to process the query.  You must also make sure that you have enough worker slots available for them. If any of the tasks are waiting to be assigned, the whole query will wait before it starts executing stages. It will fail after a timeout if worker slots do not become available.\n",
    "MSQ reads the segment files directly from Deep Storage which is also typically slower than the local storage that Historicals use and there is no caching of segment files. It has the advantage that it can access data that is not cached in the historicals but still exists in Deep Storage. So it is best for longer running queries on broader or older timeframes.\n",
    "\n",
    "As queries are running you can see the `query_controller` and `query_worker` tasks being spawned for each query in the [tasks view of the Druid Console](http://localhost:8888/unified-console.html#tasks).\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "id": "ab3a31ef-45aa-41e3-9708-4e2ec0488600",
   "metadata": {},
   "source": [
    "### Sort merge joins in MSQ\n",
    "SortMerge joins execute by hash distributing both sides of the join among all the `query_workers` using the join columns to calculate the hash. The end result is that the rows from each of the tables that correspond to a given join column value will end up on the same worker:\n",
    "![](assets/msq-sortmerge.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "554204b6-df63-4c14-96e8-4341d1b7fa8f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#MSQ async query using sortMerge\n",
    "print(measure_async_query(sql,10,joinAlgorithm='sortMerge'))\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "88a0f948-3a9e-405b-a305-95fed27db6ca",
   "metadata": {},
   "source": [
    "In this case, sortMerge joins did not help which makes sense. A broadcast of 4000 `users` rows to the workers is much faster than having to redistribute both `clicks` and `users`. But if the two tables are large, a broadcast of one of them will not be possible because it would not fit in the Heap of the `query_workers`. \n",
    "\n",
    "Even if the workers had more memory, think about the work that each method needs to do:\n",
    "- For broadcast - you can estimate the work as:\n",
    "    - number of query workers used multiplied by the number of rows being broadcast\n",
    "    - the first table in the join doesn't need to move\n",
    "- For sortMerge - work is related to the total number of rows:\n",
    "    - all rows from both tables will be shuffled once\n",
    "\n",
    "\n",
    "Try the following query that joins `clicks` with `clicks` in using `broadcast` and `sortMerge` algorithms to see this effect."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "844536e5-9159-4325-be4b-adcb13974bff",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = '''\n",
    "SELECT c.__time, c.user_id, c.client_ip, MAX(c2.__time) prior_time_in_session, count(*) prior_events_in_session \n",
    "FROM clicks c\n",
    "     INNER JOIN \n",
    "      clicks c2 ON c.user_id=c2.user_id AND c.client_ip = c2.client_ip\n",
    "WHERE c.__time > (CURRENT_TIMESTAMP  - INTERVAL '1' DAY) AND\n",
    "      c2.__time > (CURRENT_TIMESTAMP - INTERVAL '2' DAY) AND\n",
    "      c2.__time < c.__time\n",
    "GROUP BY 1,2,3\n",
    "LIMIT 10\n",
    "'''\n",
    "try:\n",
    "    result = sql_client.async_sql(sql)\n",
    "    result.wait_until_done()\n",
    "    result.rows\n",
    "except Exception as ex:\n",
    "    print(f\"Error: {ex}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d6cded7c-2019-40d7-a7dd-0ebeb5d5a134",
   "metadata": {},
   "source": [
    "As expected, the query fails because it cannot broadcast that much data to the workers. As the suggestion in the error indicates, the next run of the same query uses `sortMerge` for the join algorithm which does not have such limits. \n",
    "\n",
    "It takes a couple of minutes to complete, while it is running you can see its progress by attaching to it in the [Druid Console](http://localhost:8888/unified-console.html#workbench):\n",
    "\n",
    "![](assets/msq-attach.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9c0cc654-1c67-45e7-887f-c46f363580a7",
   "metadata": {},
   "outputs": [],
   "source": [
    "req = sql_client.sql_request(sql)\n",
    "req.add_context(\"sqlJoinAlgorithm\", 'sortMerge')\n",
    "req.add_context(\"maxNumTasks\", 3)\n",
    "result = sql_client.async_sql(req)\n",
    "result.wait_until_done()\n",
    "result=pandas.json_normalize(result.rows)\n",
    "print_timestamp()\n",
    "result"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "03a60d03-a8df-432e-85de-e9ffaf1075f2",
   "metadata": {},
   "source": [
    "## UNION ALL queries\n",
    "In some scenarios, UNION ALL queries can be used to bring data from different tables together instead of doing a join.\n",
    "There are two scenarios where [UNION ALL is supported](https://druid.apache.org/docs/latest/querying/sql#union-all).\n",
    "- Broker UNION ALL brings together the results of two subqueries. Each subquery is processed in its own scatter/gather process, the results of each subquery are returned to the Broker and the Broker returns the merged results to the client.\n",
    "- Historical/Peon UNION ALL, takes segments from two different tables and applies the same processing in all Data Servers including Historicals and Streaming Ingestion tasks. The Broker finds the relevant segments from the UNION ALL tables and submits the same query request to the data servers that own them, essentially treating multiple tables as if they were a single table. This means that the same column names must be selected from both of the tables, which can be limiting but in the rest of the notebook we'll show how it can be a powerful tool for querying multiple datasources together and how they can effectively execute joins under certain conditions.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "730c39f5-4228-48ea-a9a9-03cf19131b9f",
   "metadata": {},
   "source": [
    "### Generate `sales` and `sales_forecast` data\n",
    "In the retail industry, it is common to have a forecast of how many units you will sell in some period of time. Sales data is then aggregated to the same level as the forecast to measure progress against it. The forecast data is necessarily generated before the sales data, it wouldn't be a forecast otherwise. This creates a situation where pre-joining the data is not viable and the datasets can both be large. In this example the forecast data isn't very big but it will help illustrate how UNION ALL can be used to resolve this type of query.\n",
    "\n",
    "The next cell generates detailed `sales data` from clickstream `purchase` events and corresponding `forecast data` per country per product per day. We'll then ingest it in slightly different ways to show different UNION ALL strategies. The forecast values are somewhat random and unrealistic, so don't try to make sense of them. Instead focus on the ingestion design and query pattern."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "20204185-47af-4f1f-a965-9a759a0703a9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# simulate forecast for last 2 weeks \n",
    "gen_hours=24*14  \n",
    "gen_now = datetime.now().replace(hour=0,minute=0,second=0) - timedelta(hours=gen_hours)\n",
    "gen_start_time = gen_now.strftime(\"%Y-%m-%d %H:%M:%S\")\n",
    "\n",
    "headers = {\n",
    "  'Content-Type': 'application/json'\n",
    "}\n",
    "\n",
    "datagen_request = {\n",
    "    \"name\": \"forecast\",\n",
    "    \"target\": { \"type\": \"file\", \"path\":\"forecast.json\"  },\n",
    "    \"config_file\": \"clickstream/sales_forecast.json\", \n",
    "    \"time\": f\"{gen_hours}h\",\n",
    "    \"concurrency\":1040,\n",
    "    \"time_type\":gen_start_time\n",
    "}\n",
    "datagen.post(\"/start\", json.dumps(datagen_request), headers=headers)\n",
    "wait_for_datagen(\"forecast\")\n",
    "print_timestamp(\"after forecast data generation\")\n",
    "\n",
    "datagen_request = {\n",
    "    \"name\": \"clicks\",\n",
    "    \"target\": { \"type\": \"file\", \"path\":\"clicks.json\"  },\n",
    "    \"config_file\": \"clickstream/clickstream.json\", \n",
    "    \"time\": f\"{gen_hours}h\",\n",
    "    \"concurrency\":20,\n",
    "    \"time_type\":gen_start_time\n",
    "}\n",
    "datagen.post(\"/start\", json.dumps(datagen_request), headers=headers)\n",
    "wait_for_datagen(\"clicks\")\n",
    "print_timestamp(\"after sales data generation\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5a76c614-556a-427f-8398-cb9e1eeb6254",
   "metadata": {},
   "source": [
    "### Natural ingestion\n",
    "For the first two query patterns, we ingest the data mostly \"as is\". \n",
    "\n",
    "Run the following cell to:\n",
    "- ingest \"sales\" from clickstream data filtering for only \"event_type\"=`purchase` and adds a \"quantity\" constant set to 1 so we can aggregate sales quantity.\n",
    "- ingest \"sales_forecast\" by adding an \"event_type\" of `forecast` and renaming the source \"forecast_quantity\" to \"quantity\".\n",
    "\n",
    "The \"event_type\" is used to distinguish between `purchase` and `forecast` values in one UNION ALL scenario below. The same name for the \"quantity\" column enables the UNION ALL technique where all the columns selected from the unioned tables must be exactly the same."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1a288a4c-78a2-400c-81cb-b37d756bfbd4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# ingest sales from click data by \n",
    "# - filtering source for purchase events with \"event_type\"='purchase'\n",
    "# - adding a column for \"quantity\" with a constant value of 1\n",
    "sql='''\n",
    "REPLACE INTO \"sales\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"http://datagen:9999/file/clicks.json\"]}',\n",
    "    '{\"type\":\"json\"}'\n",
    "  )\n",
    ") EXTEND (\"time\" VARCHAR, \"user_id\" VARCHAR, \"event_type\" VARCHAR, \"client_ip\" VARCHAR, \"client_device\" VARCHAR, \"client_lang\" VARCHAR, \"client_country\" VARCHAR, \"referrer\" VARCHAR, \"keyword\" VARCHAR, \"product\" VARCHAR))\n",
    "SELECT\n",
    "  TIME_PARSE(\"time\") AS \"__time\",\n",
    "  \"user_id\",\n",
    "  \"event_type\",\n",
    "  \"client_country\",\n",
    "  \"product\",\n",
    "  1 as \"quantity\"\n",
    "FROM \"ext\"\n",
    "WHERE \"event_type\"='purchase'  \n",
    "PARTITIONED BY DAY\n",
    "'''\n",
    "display.run_task(sql)\n",
    "print_timestamp()\n",
    "\n",
    "# ingest forecast data with a couple of adjustments\n",
    "# - add column expression 'forecast' as event_type, such that we can distinguish between purchases and forecasts when using UNION ALL\n",
    "# - change the name of the \"forecast_quantity\" column to \"quantity\" so that columns names match when using UNION ALL\n",
    "sql='''\n",
    "REPLACE INTO \"sales_forecast\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"http://datagen:9999/file/forecast.json\"]}',\n",
    "    '{\"type\":\"json\"}'\n",
    "  )\n",
    ") EXTEND (\"time\" VARCHAR, \"client_country\" VARCHAR, \"product\" VARCHAR, \"forecast_quantity\" INTEGER))\n",
    "SELECT\n",
    "  TIME_PARSE(\"time\") AS \"__time\",\n",
    "  'forecast' as \"event_type\",\n",
    "  \"client_country\",\n",
    "  \"product\",\n",
    "  \"forecast_quantity\" as \"quantity\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY DAY\n",
    "'''\n",
    "display.run_task(sql)\n",
    "print_timestamp()\n",
    "\n",
    "\n",
    "# make sure both tables are available before moving on\n",
    "sql_client.wait_until_ready('sales')\n",
    "sql_client.wait_until_ready('sales_forecast')\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fca62939-b82c-4a95-bab8-906c99bfb388",
   "metadata": {},
   "source": [
    "### Independent queries where the application joins results"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2a00800c-f0bf-4d7a-8141-fa04f0569d9c",
   "metadata": {},
   "source": [
    "In order to compare forecast to sales, we will need to aggregate both at the same time granularity and with the same dimensions. \n",
    "\n",
    "One way to resolve this, is to issue two different queries, one provides the appropriate forecast aggregation by week and the other the corresponding aggregation of sales also by week. \n",
    "\n",
    "In the interest of keeping the results short, we filter for just 3 products and one country. \n",
    "\n",
    "The join of the two resultsets is done here using pandas dataframes which run on the jupyter kernel, so this is the application resolving the join:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fbbf0688-d0ea-4e24-89dd-154dfbb8cfbc",
   "metadata": {},
   "outputs": [],
   "source": [
    "# we can read data independently from each source, and join it in python\n",
    "start_at = print_timestamp()\n",
    "sql = '''\n",
    "    SELECT \n",
    "       TIME_FLOOR(__time, 'P1W') week_start, \n",
    "       \"client_country\",\n",
    "       \"product\",\n",
    "       SUM(\"quantity\") \"sales_quantity\"\n",
    "    FROM \"sales\"\n",
    "    WHERE __time > TIME_FLOOR( CURRENT_TIMESTAMP, 'P1D') - INTERVAL '7' DAY\n",
    "    AND \"product\" in ('Slinky', 'Magic 8-ball', 'Pet rock')\n",
    "    AND \"client_country\" = 'Japan'\n",
    "    GROUP BY 1,2,3\n",
    "'''\n",
    "sales_data = pandas.json_normalize(sql_client.sql(sql)).set_index(['week_start','client_country','product'])\n",
    "\n",
    "sql = '''\n",
    "    SELECT \n",
    "       TIME_FLOOR(__time, 'P1W') week_start, \n",
    "       \"client_country\",\n",
    "       \"product\",\n",
    "       SUM(\"quantity\") \"forecast_quantity\"\n",
    "    FROM \"sales_forecast\"\n",
    "    WHERE __time > TIME_FLOOR( CURRENT_TIMESTAMP, 'P1D') - INTERVAL '7' DAY\n",
    "    AND \"product\" in ('Slinky', 'Magic 8-ball', 'Pet rock')\n",
    "    AND \"client_country\" = 'Japan'\n",
    "    GROUP BY 1,2,3\n",
    "'''\n",
    "sales_forecast_data = pandas.json_normalize(sql_client.sql(sql)).set_index(['week_start','client_country','product'])\n",
    "\n",
    "# we join it here on the \"application\" side:\n",
    "result = sales_data.join(sales_forecast_data, on=['week_start','client_country','product'])\n",
    "end_at=print_timestamp()\n",
    "print(f\"Duration: {(end_at-start_at).total_seconds()*1000} ms \")\n",
    "result"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c0441a7f-4a96-4ffe-9f19-11ecb6f06b22",
   "metadata": {},
   "source": [
    "### Single query with UNION ALL - sales and forecast in different rows\n",
    "Given that both \"sales\" and \"sales_forecast\" tables have the same set of columns for the query, you can use a UNION ALL to process both aggregations in a single pass and return the results sorted such that the corresponding sales \"quantity\" and forecast \"quantity\" are in consecutive for each country, product combination.\n",
    "\n",
    "Notice that the UNION ALL selects the same columns from both tables. This is a requirement for this query pattern. If you use different projections for each table, you will get an error.\n",
    "\n",
    "The outer query with specifies the aggregation which will be executed in parallel on all segments selected from both tables in the UNION ALL. This includes the filters which will be used by the Broker for pruning which segments of each table are relevant as well as filter processing in the Historicals using column dictionaries and indexes. Since the columns are the same, the same set of historical processing instructions can be applied to segments of either table.\n",
    "\n",
    "Try it out:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "133c69c4-c881-49f4-8a89-0faeecd6a068",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = '''\n",
    "    SELECT \n",
    "       TIME_FLOOR(__time, 'P1W') week_start, \n",
    "       \"client_country\",\n",
    "       \"product\",\n",
    "       \"event_type\",\n",
    "       SUM(\"quantity\") \"quantity\"\n",
    "    FROM \n",
    "    (\n",
    "        SELECT __time, \"client_country\", \"product\", \"event_type\", \"quantity\" FROM \"sales\"\n",
    "    UNION ALL\n",
    "        SELECT __time, \"client_country\", \"product\", \"event_type\", \"quantity\" FROM \"sales_forecast\"\n",
    "    )\n",
    "    WHERE  __time > TIME_FLOOR( CURRENT_TIMESTAMP, 'P1D') - INTERVAL '7' DAY\n",
    "    AND \"product\" in ('Slinky', 'Magic 8-ball', 'Pet rock')\n",
    "    AND \"client_country\" = 'Japan'\n",
    "    GROUP BY 1,2,3,4\n",
    "    ORDER BY 1,2,3,4\n",
    "'''\n",
    "\n",
    "result=pandas.json_normalize(sql_client.sql(sql))\n",
    "print_timestamp()\n",
    "result"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e67a3e29-5822-4e1d-8c17-7d5382719d36",
   "metadata": {},
   "source": [
    "Your specific results will vary because the data generator is somewhat random. In my results I can see that some country,product combinations do not have a forecast, so there is only a purchase row. An application processing these results would need to keep that in mind. The same would be true in sales if, for a particular period, there are no sales of a given product/country combo."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "681ee0bb-6f4c-4678-bc09-b6ca4c51669d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# measure multiple query runs\n",
    "print(measure_query(sql, 20))\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a8df2973-0e41-45e8-9dd6-7b611406cee1",
   "metadata": {},
   "source": [
    "### Single query UNION ALL - sales and forecast \"joined\" through aggregation \n",
    "Another form of using the UNION ALL can produce the fully joined results, but it will require that the data be ingested with some changes:\n",
    "- add a NULL valued column called \"forecast_quantity\" to the sales table\n",
    "- add a NULL valued column called \"quantity\" to the forecast table\n",
    "\n",
    "This allows you to select both the sales \"quantity\" and \"forecast_quantity\" in the unioned tables such that resulting aggregate rows contain both. The end result is effectively a join of the two data sets through aggregated to the same granularity.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3e565a87-8476-433f-9cd9-22f500348c40",
   "metadata": {},
   "outputs": [],
   "source": [
    "# ingest sales from click data, adding a NULL valued column called \"forecast_quantity\"\n",
    "sql='''\n",
    "REPLACE INTO \"sales_u\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"http://datagen:9999/file/clicks.json\"]}',\n",
    "    '{\"type\":\"json\"}'\n",
    "  )\n",
    ") EXTEND (\"time\" VARCHAR, \"user_id\" VARCHAR, \"event_type\" VARCHAR, \"client_ip\" VARCHAR, \"client_device\" VARCHAR, \"client_lang\" VARCHAR, \"client_country\" VARCHAR, \"referrer\" VARCHAR, \"keyword\" VARCHAR, \"product\" VARCHAR))\n",
    "SELECT\n",
    "  TIME_PARSE(\"time\") AS \"__time\",\n",
    "  \"user_id\",\n",
    "  \"event_type\",\n",
    "  \"client_country\",\n",
    "  \"product\",\n",
    "  1 as \"quantity\",\n",
    "  NULL as \"forecast_quantity\"\n",
    "FROM \"ext\"\n",
    "WHERE \"event_type\"='purchase'  \n",
    "PARTITIONED BY DAY\n",
    "'''\n",
    "display.run_task(sql)\n",
    "print_timestamp()\n",
    "\n",
    "\n",
    "# ingest sales forecast data, adding a NULL valued column called \"quantity\"\n",
    "sql='''\n",
    "REPLACE INTO \"sales_forecast_u\" OVERWRITE ALL\n",
    "WITH \"ext\" AS (SELECT *\n",
    "FROM TABLE(\n",
    "  EXTERN(\n",
    "    '{\"type\":\"http\",\"uris\":[\"http://datagen:9999/file/forecast.json\"]}',\n",
    "    '{\"type\":\"json\"}'\n",
    "  )\n",
    ") EXTEND (\"time\" VARCHAR, \"client_country\" VARCHAR, \"product\" VARCHAR, \"forecast_quantity\" INTEGER))\n",
    "SELECT\n",
    "  TIME_PARSE(\"time\") AS \"__time\",\n",
    "  \"client_country\",\n",
    "  \"product\",\n",
    "  NULL as \"quantity\",\n",
    "  \"forecast_quantity\"\n",
    "FROM \"ext\"\n",
    "PARTITIONED BY DAY\n",
    "'''\n",
    "display.run_task(sql)\n",
    "print_timestamp()\n",
    "\n",
    "\n",
    "# make sure both tables are available before moving on\n",
    "sql_client.wait_until_ready('sales_u')\n",
    "sql_client.wait_until_ready('sales_forecast_u')\n",
    "print_timestamp()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ba54187-5791-455d-a3c4-40cd827c28fa",
   "metadata": {},
   "source": [
    "This query pattern works by calculating aggregation in parallel for \"sales\" segments and \"sales_forecast\" segments in the data servers. After initial segment level aggregation, the results from \"sales\" segments will have a NULL SUM for \"forecast_quantity\" and results for \"sales_forecast\" segments will have NULL on the SUM of \"quantity\" column. As these results are merged, aggregations of NULL and non-NULL values for each of the columns will result in a single row for each week,country,product combo that has both the \"quantity\" and \"forecast_quantity\". \n",
    "\n",
    "Even if a given product/country does not have a forecast or doesn't have sales in a particular period, the resulting row will still appear as long as it exists in either of the tables.\n",
    "\n",
    "Try it out:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "71e0e385-d178-4298-9e4f-e96afb66bbc2",
   "metadata": {},
   "outputs": [],
   "source": [
    "sql = '''\n",
    "    SELECT \n",
    "       TIME_FLOOR(__time, 'P1W') week_start, \n",
    "       \"client_country\",\n",
    "       \"product\",\n",
    "       SUM(\"quantity\") as \"sales_quantity\",\n",
    "       SUM(\"forecast_quantity\") as \"forecast_quantity\"\n",
    "    FROM \n",
    "    (\n",
    "        SELECT __time, \"client_country\", \"product\", \"quantity\", \"forecast_quantity\" FROM \"sales_u\"\n",
    "    UNION ALL\n",
    "        SELECT __time, \"client_country\", \"product\", \"quantity\", \"forecast_quantity\" FROM \"sales_forecast_u\"\n",
    "    )\n",
    "    WHERE __time > TIME_FLOOR( CURRENT_TIMESTAMP, 'P1D') - INTERVAL '7' DAY\n",
    "    AND \"product\" in ('Slinky', 'Magic 8-ball', 'Pet rock')\n",
    "    AND \"client_country\" = 'Japan'\n",
    "    GROUP BY 1,2,3\n",
    "    ORDER BY 1,2,3\n",
    "'''\n",
    "\n",
    "result=pandas.json_normalize(sql_client.sql(sql))\n",
    "print_timestamp()\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a63791fd-b873-47fa-b58f-86c67a8c0d86",
   "metadata": {},
   "outputs": [],
   "source": [
    "# measure multiple query runs\n",
    "print(measure_query(sql, 100))\n",
    "print_timestamp()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44738d6d-cec2-40ad-aaba-998c758c63f4",
   "metadata": {},
   "source": [
    "## Clean up\n",
    "\n",
    "Run the following cell to remove the tables created throughout this notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8082b545-ba7f-4ede-bb6e-2a6dd62ba0d8",
   "metadata": {},
   "outputs": [],
   "source": [
    "druid.datasources.drop(\"clicks\")\n",
    "druid.datasources.drop(\"users\")\n",
    "druid.datasources.drop(\"clicks_enhanced\")\n",
    "druid.datasources.drop(\"sales\")\n",
    "druid.datasources.drop(\"sales_forecast\")\n",
    "druid.datasources.drop(\"sales_u\")\n",
    "druid.datasources.drop(\"sales_forecast_u\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "54b8d5fe-ba85-4b5b-9669-0dd47dfbccd1",
   "metadata": {},
   "source": [
    "## Summary\n",
    "\n",
    "Druid can process data join operations at query time in different ways:\n",
    "<br>\n",
    "- In the scatter/gather model:\n",
    "    * joined tables are first scanned and the result is broadcast to all data servers to process the join with the first table\n",
    "    * joining a table to subqueries means the subqueries are processed as separate queries, and their results are then broadcast to join with the first table\n",
    "    * order of the joined tables is important, the larger table should be listed first such that it drives parallelization of the join\n",
    "    * there is a limit to how much data can be broadcast set by either `maxSubQueryRows` or `maxSubQueryBytes` limits\n",
    "    * you can raise the subquery limits but it comes at the cost of more Heap memory used on the Broker, Historicals, and Streaming Tasks to resolve the query and more data transmitted from the Broker to all the others involved in the query\n",
    "<br>\n",
    "<br>\n",
    "- In the MSQ engine query model:\n",
    "    * meant for longer running queries with more complex joins\n",
    "    * queries are generally slower because:\n",
    "        * they need to spawn workers in order to run\n",
    "        * they read the segment data from Deep Storage\n",
    "    * joins can use either broadcast or sortMerge join algorithms\n",
    "    * with sortMerge joins, large fact to fact joins are possible \n",
    "<br>\n",
    "<br>\n",
    "- UNION ALL can be an powerful alternative to large joins\n",
    "    * for correctness, queries must aggregate to the coarsest time granularity of the unioned tables\n",
    "    * all columns involved in the query need to exist on all unioned tables\n",
    "    * by adding NULL valued columns with the same name to the other table(s) UNION ALL aggregation can effectively do fact to fact joins    "
   ]
  }
 ],
 "metadata": {
  "execution": {
   "allow_errors": true,
   "timeout": 300
  },
  "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"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}