{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# La carga de las tres de la mañana\n",
    "\n",
    "Traer solo lo que llegó desde la última vez, sin duplicar nada y sin perder lo que llegó tarde. Con los dos errores que salen siempre.\n",
    "\n",
    "Cuaderno de soluciones del capítulo 16 de **SQL desde cero**, de Miss Yera.\n",
    "\n",
    "Corre de arriba abajo. Si lo abres en Google Colab no necesitas instalar nada.\n",
    "\n",
    "Capítulo completo: https://missyera.com/guias/sql-desde-cero/cargar-datos-cada-dia/\n",
    "\n",
    "Este es el cuaderno de **soluciones**. Trae el código de cada ejercicio, la\n",
    "explicación de la trampa y la respuesta del quiz. Si vienes del cuaderno de\n",
    "práctica sin haberlo intentado, vuelve 🙂"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Antes de empezar\n",
    "\n",
    "Se baja la base y se deja lista una función `q()` que corre las consultas."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sqlite3\n",
    "import urllib.request\n",
    "\n",
    "import pandas as pd\n",
    "\n",
    "urllib.request.urlretrieve(\"https://missyera.com/static/datasets/tienda.db\", \"tienda.db\")\n",
    "con = sqlite3.connect(\"tienda.db\")\n",
    "\n",
    "def q(sql):\n",
    "    \"\"\"Corre las sentencias del bloque y devuelve la ultima como tabla.\n",
    "\n",
    "    Parte por sentencias igual que la consola de SQLite, porque un bloque puede\n",
    "    traer varias y un CREATE TRIGGER lleva punto y coma dentro de su cuerpo.\n",
    "    \"\"\"\n",
    "    resultado = None\n",
    "    trozo = \"\"\n",
    "    for linea in sql.splitlines(keepends=True):\n",
    "        trozo += linea\n",
    "        if sqlite3.complete_statement(trozo):\n",
    "            if trozo.strip():\n",
    "                cur = con.execute(trozo.strip())\n",
    "                resultado = (pd.DataFrame(cur.fetchall(),\n",
    "                                          columns=[d[0] for d in cur.description])\n",
    "                             if cur.description else None)\n",
    "            trozo = \"\"\n",
    "    if trozo.strip():\n",
    "        cur = con.execute(trozo.strip())\n",
    "        resultado = (pd.DataFrame(cur.fetchall(),\n",
    "                                  columns=[d[0] for d in cur.description])\n",
    "                     if cur.description else None)\n",
    "    con.commit()\n",
    "    return resultado\n",
    "\n",
    "q(\"SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "En el capítulo 15 armamos el almacén de una vez.\n",
    "Ahora la parte que se repite: **mañana llegan los pedidos de hoy**,\n",
    "y pasado los de mañana.\n",
    "\n",
    "Las dos formas de hacerlo son borrar todo y volver a cargar, o traer solo lo\n",
    "nuevo. La primera es más simple y aguanta bastante más de lo que la gente cree;\n",
    "la segunda es la que hace falta cuando la tabla ya no cabe en la ventana de la\n",
    "madrugada.\n",
    "\n",
    "Preparo la base con una columna que casi ninguna tabla de sistema trae de\n",
    "fábrica y que es el centro del capítulo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "ALTER TABLE pedidos ADD COLUMN cargado_en TEXT;\n",
    "UPDATE pedidos SET cargado_en = fecha;\n",
    "\n",
    "SELECT count(*) AS pedidos, min(fecha) AS desde, max(fecha) AS hasta FROM pedidos;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Novecientos pedidos, año y medio. De momento `cargado_en` es igual\n",
    "a `fecha`, o sea que hago como si cada pedido hubiera entrado el\n",
    "mismo día que se hizo. Dentro de un rato dejará de ser verdad, y ahí está el\n",
    "capítulo entero 🙂"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## La marca de agua"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "La idea es de una línea: *guarda hasta dónde cargaste, y la próxima vez\n",
    "trae lo que venga después*. Eso que guardas se llama marca de agua, o\n",
    "*watermark* si estás leyendo documentación.\n",
    "\n",
    "Lo bonito es que no hace falta guardarla en ningún sitio: se le pregunta a la\n",
    "tabla de destino."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "CREATE TABLE ventas_dw (\n",
    "    id_pedido  INTEGER PRIMARY KEY,\n",
    "    fecha      TEXT NOT NULL,\n",
    "    canal      TEXT,\n",
    "    monto      REAL,\n",
    "    cargado_en TEXT\n",
    ");\n",
    "\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "WHERE cargado_en > (SELECT max(cargado_en) FROM ventas_dw);\n",
    "\n",
    "SELECT count(*) AS filas_cargadas FROM ventas_dw;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cero 😐\n",
    "\n",
    "Y no falló nada. La tabla estaba vacía, `max(cargado_en)` sobre una\n",
    "tabla vacía devuelve `NULL`, y cualquier comparación contra\n",
    "`NULL` da `NULL`, que no es verdadero. El\n",
    "`WHERE` no dejó pasar ni una fila.\n",
    "\n",
    "Es la lección de los nulos del capítulo 4 llegando a la\n",
    "peor hora posible: la primera carga de la historia, la que nadie mira porque\n",
    "\"acaba de empezar\". Se arregla con un `coalesce`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "WHERE cargado_en > coalesce((SELECT max(cargado_en) FROM ventas_dw), '0000-00-00')\n",
    "  AND cargado_en < '2026-06-20';\n",
    "\n",
    "SELECT count(*) AS filas, max(cargado_en) AS marca FROM ventas_dw;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "890 filas y la marca en el 19 de junio. El `AND` del final está\n",
    "para simular que hoy es 20 de junio y quedan cuatro días por cargar."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Correrla otra vez, que es lo que va a pasar"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ahora la carga de la noche siguiente. Y voy a escribirla con el error que se\n",
    "escribe siempre, que es usar `>=` en vez de `>` \"por\n",
    "si acaso\":"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Esto revienta a propósito.** Se ejecuta dentro de un `try` para que puedas seguir con \"ejecutar todo\" y aun así ver la queja."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    q(\"\"\"\n",
    "    INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "    SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "    WHERE cargado_en >= (SELECT max(cargado_en) FROM ventas_dw);\n",
    "    \"\"\")\n",
    "except Exception as e:\n",
    "    print(f'{type(e).__name__}: {e}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Y la queja que tiene que salir es esta:\n",
    "\n",
    "```\n",
    "IntegrityError: UNIQUE constraint failed: ventas_dw.id_pedido\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "El `>=` vuelve a traer el día que ya estaba cargado, y la clave\n",
    "primaria lo caza. Y acá hay que decir algo importante: **que reviente es\n",
    "la buena noticia**.\n",
    "\n",
    "Sin esa clave primaria, esas filas habrían entrado dos veces, la carga habría\n",
    "terminado en verde, y la venta del 19 de junio saldría duplicada en el reporte\n",
    "del lunes sin que nadie lo note. La restricción del capítulo\n",
    "11 está haciendo exactamente su trabajo 💛"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Que se pueda correr dos veces: idempotencia"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "La palabra suena a examen y la idea es simple: **correr la carga dos\n",
    "veces tiene que dejar la tabla igual que correrla una**.\n",
    "\n",
    "Hace falta porque las cargas se repiten todo el tiempo. Se cayó a mitad y la\n",
    "relanzas. Alguien la lanzó a mano sin saber que ya había corrido. El servidor se\n",
    "reinició y el planificador la disparó otra vez. Si repetir duplica, cualquiera\n",
    "de esas tres te ensucia el almacén.\n",
    "\n",
    "La forma es el upsert del capítulo 12:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "WHERE cargado_en >= (SELECT max(cargado_en) FROM ventas_dw)\n",
    "ON CONFLICT (id_pedido) DO UPDATE\n",
    "SET fecha = excluded.fecha, canal = excluded.canal, monto = excluded.monto,\n",
    "    cargado_en = excluded.cargado_en;\n",
    "\n",
    "SELECT count(*) AS filas, max(cargado_en) AS marca,\n",
    "       round(sum(monto), 2) AS venta FROM ventas_dw;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Las 900. Y ahora la prueba, que es correr lo mismo otra vez:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "WHERE cargado_en >= (SELECT max(cargado_en) FROM ventas_dw)\n",
    "ON CONFLICT (id_pedido) DO UPDATE\n",
    "SET fecha = excluded.fecha, canal = excluded.canal, monto = excluded.monto,\n",
    "    cargado_en = excluded.cargado_en;\n",
    "\n",
    "SELECT count(*) AS filas, round(sum(monto), 2) AS venta FROM ventas_dw;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Mismo número de filas y misma venta. Eso es idempotente, y es lo que te deja\n",
    "dormir cuando la carga falla a las tres de la mañana: relanzas y ya 🌙"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Los dos relojes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Acá viene lo que más datos pierde en el mundo real, y es tan silencioso como\n",
    "el JOIN del capítulo anterior.\n",
    "\n",
    "Entra un pedido con fecha del 10 de junio, que se cargó hoy 25 porque el\n",
    "marketplace lo mandó tarde:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "INSERT INTO pedidos (id, id_cliente, fecha, canal, monto, cargado_en)\n",
    "VALUES (99001, 7, '2026-06-10', 'WhatsApp', 480.5, '2026-06-25');\n",
    "\n",
    "SELECT (SELECT count(*) FROM pedidos)   AS en_el_origen,\n",
    "       (SELECT count(*) FROM ventas_dw) AS en_el_almacen,\n",
    "       (SELECT max(fecha) FROM ventas_dw) AS marca_por_fecha,\n",
    "       (SELECT max(cargado_en) FROM ventas_dw) AS marca_por_carga;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "901 en el origen, 900 en el almacén. Corre la carga usando la fecha del\n",
    "pedido, que es lo que hace casi todo el mundo:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "WHERE fecha >= (SELECT max(fecha) FROM ventas_dw)\n",
    "ON CONFLICT (id_pedido) DO UPDATE\n",
    "SET fecha = excluded.fecha, monto = excluded.monto, cargado_en = excluded.cargado_en;\n",
    "\n",
    "SELECT count(*) AS en_el_almacen FROM ventas_dw;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sigue en 900. La venta de 480,50 soles existe, entró bien, está en el origen,\n",
    "y tu almacén no la va a ver **nunca**, porque su fecha es más vieja\n",
    "que la marca de agua y ese pedido ya pasó de largo.\n",
    "\n",
    "La cura es cambiar de reloj:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "WHERE cargado_en >= (SELECT max(cargado_en) FROM ventas_dw)\n",
    "ON CONFLICT (id_pedido) DO UPDATE\n",
    "SET fecha = excluded.fecha, monto = excluded.monto, cargado_en = excluded.cargado_en;\n",
    "\n",
    "SELECT (SELECT count(*) FROM pedidos)   AS en_el_origen,\n",
    "       (SELECT count(*) FROM ventas_dw) AS en_el_almacen;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Los dos relojes tienen nombre y conviene sabérselos porque salen en toda la\n",
    "documentación: la fecha del pedido es **hora del negocio**, y la de\n",
    "carga es **hora de ingesta**. Se reportan por la primera y se carga\n",
    "por la segunda 📏\n",
    "\n",
    "Y si el sistema de origen no guarda la hora de ingesta, esa es una petición\n",
    "de las que se piden el día uno, como en cualquier proyecto de datos."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## La bitácora, que contesta \"¿corrió?\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Una tabla más, y es la que te salva de la peor pregunta de un lunes, que es\n",
    "\"¿esto está actualizado?\":"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "CREATE TABLE carga_log (\n",
    "    id         INTEGER PRIMARY KEY,\n",
    "    tabla      TEXT,\n",
    "    corrio_en  TEXT,\n",
    "    desde      TEXT,\n",
    "    filas      INTEGER,\n",
    "    estado     TEXT\n",
    ");\n",
    "\n",
    "INSERT INTO carga_log (tabla, corrio_en, desde, filas, estado) VALUES\n",
    "    ('ventas_dw', '2026-06-20 03:00', '2026-06-19', 10, 'ok'),\n",
    "    ('ventas_dw', '2026-06-25 03:00', '2026-06-24',  1, 'ok'),\n",
    "    ('ventas_dw', '2026-06-26 03:00', '2026-06-25',  0, 'sin datos');\n",
    "\n",
    "SELECT corrio_en, desde, filas, estado FROM carga_log ORDER BY corrio_en;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Mira la última fila. Cero filas no es un error: puede ser un domingo sin\n",
    "ventas o puede ser que el origen dejó de mandar. Las dos se ven igual, y por eso\n",
    "se anota en vez de discutirse de memoria."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Enterarse de qué cambió en el origen, sin preguntar tabla por tabla\n",
    "\n",
    "| PostgreSQL | `replicación lógica, o una columna actualizado_en puesta por ti` |\n",
    "|---|---|\n",
    "| MySQL | `el binlog, que se suele leer con Debezium` |\n",
    "| SQL Server | `Change Data Capture, incluido en el motor` |\n",
    "| SQLite | `no tiene nada: se pone a mano con un TRIGGER` |\n",
    "\n",
    "Esto se llama captura de cambios, y lo cuento para que sepas que existe antes de que alguien te lo nombre en una reunión. Para casi todo lo que vas a hacer, una columna con la hora de ingesta y una marca de agua alcanzan de sobra, y se entienden leyéndolas."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Comprueba que lo tienes\n",
    "\n",
    "Tu carga de cada noche trae lo que llegó después de la última vez, y usa la fecha del pedido para saber cuál fue la última vez. ¿Dónde está el problema?\n",
    "\n",
    "a) En un pedido con fecha vieja que entra hoy: la carga nunca lo va a ver\n",
    "\n",
    "b) En que la fecha del pedido puede venir nula\n",
    "\n",
    "c) En que hay que ordenar por fecha antes de cargar\n",
    "\n",
    "d) En nada, es la forma correcta\n",
    "\n",
    "---\n",
    "\n",
    "**La correcta es la a.**\n",
    "\n",
    "*b)* Es un problema real y se arregla con un NOT NULL. El otro no se arregla con una restricción porque el dato no está mal: llegó tarde.\n",
    "\n",
    "*c)* El orden no cambia qué filas entran. Entran las que cumplen el WHERE.\n",
    "\n",
    "*d)* Es la forma más común y por eso la que más datos pierde. La fecha del pedido dice cuándo pasó; no dice cuándo te enteraste.\n",
    "\n",
    "La marca de agua se pone sobre la hora en que el dato ENTRÓ, no sobre la fecha del negocio. Son dos relojes distintos y confundirlos es el error más caro de una carga incremental 🧾"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Ejercicios"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. La ventana, que es más simple que la marca\n",
    "\n",
    "En vez de traer lo nuevo, borra los últimos días y\n",
    "vuélvelos a cargar."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "DELETE FROM ventas_dw WHERE cargado_en >= '2026-06-18';\n",
    "\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "WHERE cargado_en >= '2026-06-18';\n",
    "\n",
    "SELECT count(*) AS filas, round(sum(monto), 2) AS venta FROM ventas_dw;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "filas  venta\n",
    "-----  ---------\n",
    "901    533134.35\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Borrar y reinsertar una ventana de días es idempotente sin necesitar upsert,\n",
    "y se lee de un vistazo. Es lo que yo haría en un proyecto chico.\n",
    "\n",
    "Lo que hay que cuidar es que el borrado y la inserción vayan\n",
    "**en la misma transacción**. Si se cae en medio, la ventana queda\n",
    "vacía y el reporte de esa mañana enseña un agujero que no existe."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. El monto que cambió después\n",
    "\n",
    "Un pedido se corrige en el origen tres días más tarde.\n",
    "¿Llega la corrección?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "UPDATE pedidos SET monto = monto + 100 WHERE id = 99001;\n",
    "\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos WHERE id = 99001\n",
    "ON CONFLICT (id_pedido) DO UPDATE SET monto = excluded.monto;\n",
    "\n",
    "SELECT id_pedido, monto FROM ventas_dw WHERE id_pedido = 99001;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "id_pedido  monto\n",
    "---------  -----\n",
    "99001      580.5\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Llega, porque el upsert actualiza. Pero fíjate en la trampa: la traje\n",
    "diciendo `WHERE id = 99001`, o sea que yo ya sabía cuál había\n",
    "cambiado.\n",
    "\n",
    "Una carga de verdad no lo sabe. Si el sistema no toca\n",
    "`cargado_en` al corregir una fila, la corrección es invisible para la\n",
    "marca de agua. Por eso esa columna se actualiza en cada cambio y no solo al\n",
    "insertar 🧾"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. El borrado que el incremental no ve nunca\n",
    "\n",
    "El pedido se anula en el origen. Mira qué pasa."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "DELETE FROM pedidos WHERE id = 99001;\n",
    "\n",
    "SELECT (SELECT count(*) FROM pedidos)   AS en_el_origen,\n",
    "       (SELECT count(*) FROM ventas_dw) AS en_el_almacen,\n",
    "       (SELECT count(*) FROM ventas_dw v\n",
    "        WHERE NOT EXISTS (SELECT 1 FROM pedidos p WHERE p.id = v.id_pedido)) AS fantasmas;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "en_el_origen  en_el_almacen  fantasmas\n",
    "------------  -------------  ---------\n",
    "900           901            1\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Un fantasma: una venta que el almacén sigue contando y que en el origen ya no\n",
    "existe. Ninguna carga incremental lo puede detectar sola, porque las filas\n",
    "borradas no aparecen en ningún `SELECT`.\n",
    "\n",
    "Las tres salidas: pedirle al sistema que no borre y marque como anulado, que\n",
    "es lo mejor; correr una comparación completa de vez en cuando, que es esta misma\n",
    "consulta; o recargar todo de tanto en tanto. La primera se pide, las otras dos\n",
    "se programan."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Que la bitácora te hable\n",
    "\n",
    "Escribe la consulta que quieres que alguien mire cada\n",
    "mañana."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "SELECT corrio_en, filas, estado FROM carga_log\n",
    "WHERE filas = 0 OR estado <> 'ok';\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "corrio_en         filas  estado\n",
    "----------------  -----  ---------\n",
    "2026-06-26 03:00  0      sin datos\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Una consulta de dos líneas. Si devuelve algo, alguien mira; si no devuelve\n",
    "nada, nadie tiene que hacer nada.\n",
    "\n",
    "Eso último es lo que la hace útil. Un tablero de cargas que hay que abrir y\n",
    "revisar deja de abrirse a las tres semanas; una consulta que solo habla cuando\n",
    "hay algo que decir sobrevive años 📏"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5. El dato sucio que congela la carga para siempre\n",
    "\n",
    "Entra un pedido con fecha del año 2099, de esos que salen\n",
    "de un formulario mal llenado."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "q(\"\"\"\n",
    "INSERT INTO pedidos (id, id_cliente, fecha, canal, monto, cargado_en)\n",
    "VALUES (99002, 7, '2099-01-01', 'Web', 12.0, '2099-01-01');\n",
    "\n",
    "INSERT INTO ventas_dw (id_pedido, fecha, canal, monto, cargado_en)\n",
    "SELECT id, fecha, canal, monto, cargado_en FROM pedidos\n",
    "WHERE cargado_en > (SELECT max(cargado_en) FROM ventas_dw)\n",
    "ON CONFLICT (id_pedido) DO UPDATE SET monto = excluded.monto;\n",
    "\n",
    "SELECT max(cargado_en) AS marca_de_agua,\n",
    "       (SELECT count(*) FROM pedidos\n",
    "        WHERE cargado_en > (SELECT max(cargado_en) FROM ventas_dw)) AS lo_que_traeria_manana\n",
    "FROM ventas_dw;\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```\n",
    "marca_de_agua  lo_que_traeria_manana\n",
    "-------------  ---------------------\n",
    "2099-01-01     0\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "La marca de agua se fue al 2099 y a partir de ahora **ninguna carga va\n",
    "a traer nada**. Durante setenta y tres años.\n",
    "\n",
    "Y lo peor es cómo se ve desde fuera: la carga corre todas las noches, termina\n",
    "en verde, y trae cero filas. Nadie sospecha de un proceso que no falla.\n",
    "\n",
    "La defensa es un tope: `AND cargado_en <= date('now')` en el\n",
    "`WHERE`, y una fila en la bitácora avisando de que había datos del\n",
    "futuro. Un dato imposible tiene que ser ruidoso 🚨"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 6. Completa o incremental, en tu caso\n",
    "\n",
    "Cuatro tablas. ¿Cuál cargarías entera cada noche?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**El maestro de productos, 40 filas.** Entera. Escribir una\n",
    "carga incremental para 40 filas es trabajo que se paga en errores y no ahorra\n",
    "nada.\n",
    "\n",
    "**El maestro de clientes, 120 filas que casi no cambian.**\n",
    "Entera también, y de paso te resuelve gratis el problema de los borrados del\n",
    "ejercicio 3.\n",
    "\n",
    "**Las líneas de venta, 2.682 y creciendo cada día.** Acá el\n",
    "incremental empieza a tener sentido, y aun así una tabla de este tamaño se\n",
    "recarga en segundos. La pregunta no es cuántas filas tiene hoy, es cuántas va a\n",
    "tener en tres años.\n",
    "\n",
    "**El registro de clics de la web, millones al mes.** Incremental\n",
    "sin discusión, y por día. Esa es la tabla donde la carga completa deja de caber\n",
    "en la madrugada, que es el único motivo de verdad para complicarse 💛"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Preguntas frecuentes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "¿Qué es un watermark en un pipeline de datos?La marca de hasta dónde cargaste la última vez. La siguiente carga trae solo lo que venga después, y se le pregunta a la propia tabla de destino.\n",
    "\n",
    "¿Qué es la idempotencia?Que correr algo dos veces deje el mismo resultado que correrlo una. Es lo que te deja relanzar una carga que se cayó sin pensarlo dos veces.\n",
    "\n",
    "¿Qué es una carga incremental?Traer solo lo nuevo en vez de recargar la tabla entera. Hace falta cuando la carga completa ya no cabe en la ventana de la madrugada, y no antes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "Ese era el capítulo 16 de **SQL desde cero**. El texto completo, con las salidas de cada bloque, está en https://missyera.com/guias/sql-desde-cero/cargar-datos-cada-dia/\n",
    "\n",
    "Que tengas lindo día! 🌸"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}