demo: seed via CTE injection instead of CREATE TEMPORARY VIEW

The e2e run revealed that a DDL/command ExecutePlan over the grpc-web reattach
path livelocks (an infinite ReattachExecute loop), hanging the demo at seeding.
None of the existing passing tests issue a command - only data-returning
queries - so the command path was unexercised.

Define the synthetic retail dataset as CTEs injected into every query, so every
statement sent to Spark is a data-returning SELECT (the working path). The
table picker / DESCRIBE / queries are unchanged from the user's perspective.

Co-authored-by: Isaac

Author: HyukjinKwon

demo/README.md

@@ -20,11 +20,16 @@ What it does:
 - **Runs your SQL** — `spark.sql(...).toPandas()` over the blocking SAB/Atomics
   bridge, rendered as a result grid (first 1000 rows).
 - **Ships sample data** — a synthetic retail dataset (`customers`, `products`,
-  `orders`) is seeded on load with `CREATE OR REPLACE TEMPORARY VIEW … AS SELECT
-  … FROM range(N)`. The data is deterministic (seeded `rand`/`pmod`), so every
-  session gets the identical dataset, with no warehouse writes or catalog config
-  needed — it works on any Spark Connect server. Plus four example analytics
-  queries (top products, revenue by country, monthly revenue, top customers).
+  `orders`) defined as deterministic CTEs (seeded `rand`/`pmod`) that are injected
+  into every query, so the "tables" need no DDL: every statement sent to Spark is
+  a data-returning `SELECT`. Same dataset for every visitor, no warehouse writes,
+  works on any Spark Connect server. Plus four example analytics queries (top
+  products, revenue by country, monthly revenue, top customers).
+
+  > Why CTEs and not `CREATE TEMPORARY VIEW`? Command/DDL responses currently
+  > livelock the grpc-web reattach path (an `ExecutePlan` for a command spins in
+  > `ReattachExecute`); data-returning `SELECT`s work. The CTE injection keeps the
+  > demo on the working path. See the note in `demo/index.html`.
 
 It is the same boot path as [`jupyterlite/harness.html`](../pyspark_connect_web/jupyterlite/harness.html)
 (a module Web Worker running `worker_bootstrap.js` + the `bridge.js` blocking

demo/index.html

@@ -307,49 +307,53 @@ <h2>Starting PySpark in your browser</h2>
         "        return v",
         "    return str(v)",
         "",
-        "def pcw_seed():",
-        "    # Per-session TEMPORARY VIEWs: deterministic (seeded rand/pmod), so every",
-        "    # visitor gets the identical dataset, with no warehouse writes or catalog",
-        "    # config needed - works on any Spark Connect server, and is self-contained.",
-        "    spark.sql('''CREATE OR REPLACE TEMPORARY VIEW products AS",
-        "      SELECT id AS product_id,",
-        "             element_at(array('Widget','Gadget','Gizmo','Doohickey','Sprocket','Cog','Lever','Valve'), cast(pmod(id,8) as int)+1) || ' #' || cast(id as string) AS product_name,",
-        "             element_at(array('Electronics','Home','Toys','Sports','Office'), cast(pmod(id,5) as int)+1) AS category,",
-        "             round(5 + rand(id)*195, 2) AS price",
-        "      FROM range(50)''')",
-        "    spark.sql('''CREATE OR REPLACE TEMPORARY VIEW customers AS",
-        "      SELECT id AS customer_id,",
-        "             'Customer ' || cast(id as string) AS name,",
-        "             element_at(array('US','UK','DE','FR','JP','IN','BR','CA'), cast(pmod(id,8) as int)+1) AS country,",
-        "             date_add(date'2023-01-01', cast(pmod(id*7,700) as int)) AS signup_date",
-        "      FROM range(200)''')",
-        "    spark.sql('''CREATE OR REPLACE TEMPORARY VIEW orders AS",
-        "      SELECT id AS order_id,",
-        "             cast(pmod(id*31+7, 200) as bigint) AS customer_id,",
-        "             cast(pmod(id*17+3, 50) as bigint) AS product_id,",
-        "             cast(1 + pmod(id*13, 5) as int) AS quantity,",
-        "             date_add(date'2024-01-01', cast(pmod(id*3, 365) as int)) AS order_date",
-        "      FROM range(5000)''')",
-        "    _emit({'ok': True})",
+        "# The synthetic retail dataset, defined as CTEs (deterministic seeded",
+        "# rand/pmod). We inject these into every query so the demo's 'tables' need",
+        "# no DDL: every statement sent to Spark is a data-returning SELECT. This is",
+        "# deliberate - CREATE/temp-view COMMAND responses currently livelock the",
+        "# grpc-web reattach path, while SELECT results work. Same dataset for every",
+        "# visitor, no warehouse writes, works on any Spark Connect server.",
+        "_CTES = '''",
+        "  products AS (",
+        "    SELECT id AS product_id,",
+        "           element_at(array('Widget','Gadget','Gizmo','Doohickey','Sprocket','Cog','Lever','Valve'), cast(pmod(id,8) as int)+1) || ' #' || cast(id as string) AS product_name,",
+        "           element_at(array('Electronics','Home','Toys','Sports','Office'), cast(pmod(id,5) as int)+1) AS category,",
+        "           round(5 + rand(id)*195, 2) AS price",
+        "    FROM range(50)),",
+        "  customers AS (",
+        "    SELECT id AS customer_id,",
+        "           'Customer ' || cast(id as string) AS name,",
+        "           element_at(array('US','UK','DE','FR','JP','IN','BR','CA'), cast(pmod(id,8) as int)+1) AS country,",
+        "           date_add(date'2023-01-01', cast(pmod(id*7,700) as int)) AS signup_date",
+        "    FROM range(200)),",
+        "  orders AS (",
+        "    SELECT id AS order_id,",
+        "           cast(pmod(id*31+7, 200) as bigint) AS customer_id,",
+        "           cast(pmod(id*17+3, 50) as bigint) AS product_id,",
+        "           cast(1 + pmod(id*13, 5) as int) AS quantity,",
+        "           date_add(date'2024-01-01', cast(pmod(id*3, 365) as int)) AS order_date",
+        "    FROM range(5000))",
+        "'''",
+        "",
+        "_TABLES = ['customers', 'orders', 'products']",
+        "",
+        "def _wrap(sql):",
+        "    s = sql.strip().rstrip(';').strip()",
+        "    if s[:5].lower() == 'with ':",
+        "        return 'WITH ' + _CTES + ',\\n' + s[5:]",
+        "    return 'WITH ' + _CTES + '\\n' + s",
         "",
         "def pcw_tables():",
-        "    rows = spark.sql('SHOW TABLES').collect()",
-        "    _emit([r['tableName'] for r in rows])",
+        "    _emit(_TABLES)",
         "",
         "def pcw_describe(b64):",
         "    name = base64.b64decode(b64).decode('utf-8')",
-        "    rows = spark.sql('DESCRIBE TABLE ' + name).collect()",
-        "    cols = []",
-        "    for r in rows:",
-        "        cn = r['col_name']",
-        "        if not cn or cn.startswith('#'):",
-        "            break",
-        "        cols.append([cn, r['data_type']])",
-        "    _emit(cols)",
+        "    sdf = spark.sql(_wrap('SELECT * FROM ' + name + ' LIMIT 0'))",
+        "    _emit([[f.name, f.dataType.simpleString()] for f in sdf.schema.fields])",
         "",
         "def pcw_query(b64, limit):",
         "    sql = base64.b64decode(b64).decode('utf-8')",
-        "    sdf = spark.sql(sql)",
+        "    sdf = spark.sql(_wrap(sql))",
         "    cols = list(sdf.columns)",
         "    pdf = sdf.limit(limit + 1).toPandas()",
         "    truncated = len(pdf) > limit",
@@ -569,8 +573,7 @@ <h2>Starting PySpark in your browser</h2>
           await ready;
           bootStep("connecting to Spark (" + SPARK_REMOTE + ")…", 55);
           await call(PY_BOOTSTRAP);
-          bootStep("seeding the demo retail dataset…", 80);
-          await call("pcw_seed()");
+          bootStep("loading tables…", 85);
           await renderTables();
           bootStep("ready", 100);
           $("overlay").style.display = "none";