feat(historical): offload historical_sales to S3 (Iceberg/Parquet on Dell ECS), repoint dashboard, drop Postgres copy

infra/historical/historical_postgres.sql

@@ -1,18 +0,0 @@
-DROP TABLE IF EXISTS public.historical_sales;
-CREATE TABLE public.historical_sales (
-  order_id bigint, order_date date, order_year int, region text,
-  product_category text, channel text, quantity int,
-  unit_price double precision, amount double precision
-);
-INSERT INTO public.historical_sales
-SELECT s,
-  DATE '2019-01-01' + ((s*7)%2190),
-  EXTRACT(YEAR FROM (DATE '2019-01-01' + ((s*7)%2190)))::int,
-  (ARRAY['EMEA','AMER','APAC','LATAM'])[(s%4)+1],
-  (ARRAY['Servers','Storage','Networking','Laptops','Services'])[(s%5)+1],
-  (ARRAY['Direct','Partner','Online'])[(s%3)+1],
-  (1+(s%50))::int,
-  round((100+(s%9000)*0.5)::numeric,2)::double precision,
-  round(((1+(s%50))*(100+(s%9000)*0.5))::numeric,2)::double precision
-FROM generate_series(1,25000) s;
-SELECT order_year, count(*), round(sum(amount)::numeric,0) FROM public.historical_sales GROUP BY order_year ORDER BY 1;

infra/historical/historical_s3_load.py

@@ -0,0 +1,26 @@
+import json, urllib.request
+Q="\x27"
+def run(sql):
+    req=urllib.request.Request("http://10.0.21.50:8089/v1/statement", data=sql.encode(), headers={"X-Trino-User":"admin"})
+    d=json.load(urllib.request.urlopen(req)); nu=d.get("nextUri"); err=d.get("error")
+    while nu:
+        j=json.load(urllib.request.urlopen(nu))
+        if j.get("error"): err=j["error"]
+        nu=j.get("nextUri")
+    return err
+def chunk(off,size):
+    sql=f"""INSERT INTO iceberg.s3lake.historical_sales
+SELECT order_id, order_date, year(order_date) AS order_year, region, product_category, channel, quantity, unit_price, ROUND(quantity*unit_price,2) AS amount
+FROM (
+  SELECT (s+{off}) AS order_id,
+    DATE {Q}2019-01-01{Q} + CAST(((s+{off})*7)%2190 AS bigint) * INTERVAL {Q}1{Q} DAY AS order_date,
+    element_at(ARRAY[{Q}EMEA{Q},{Q}AMER{Q},{Q}APAC{Q},{Q}LATAM{Q}], CAST((s+{off})%4 AS int)+1) AS region,
+    element_at(ARRAY[{Q}Servers{Q},{Q}Storage{Q},{Q}Networking{Q},{Q}Laptops{Q},{Q}Services{Q}], CAST((s+{off})%5 AS int)+1) AS product_category,
+    element_at(ARRAY[{Q}Direct{Q},{Q}Partner{Q},{Q}Online{Q}], CAST((s+{off})%3 AS int)+1) AS channel,
+    CAST(1+((s+{off})%50) AS int) AS quantity,
+    ROUND(100+((s+{off})%9000)*0.5,2) AS unit_price
+  FROM UNNEST(SEQUENCE(1,{size})) AS t(s)
+)"""
+    return run(sql)
+for off,size in [(0,10000),(10000,10000),(20000,5000)]:
+    e=chunk(off,size); print(f"chunk {off}:", e.get("message") if e else "ok")

infra/historical/superset_repoint_s3.py

@@ -7,18 +7,18 @@ with app.app_context():
     from superset.models.core import Database
 
     t = db.session.query(SqlaTable).filter_by(table_name="historical_sales").first()
-    pg = (
+    ice = (
         db.session.query(Database)
-        .filter(Database.database_name.like("%PostgreSQL Sales%"))
+        .filter(Database.database_name.like("%Iceberg Lakehouse%"))
         .first()
     )
-    t.database_id = pg.id
+    t.database_id = ice.id
-    t.schema = "public"
+    t.schema = "s3lake"
     db.session.commit()
     try:
         t.fetch_metadata()
         db.session.commit()
     except Exception as e:
         print("fetch_metadata warn:", str(e)[:200])
-    print("repointed dataset", t.id, "-> db", pg.id, pg.database_name,
+    print("repointed dataset", t.id, "-> db", ice.id, ice.database_name,
           "schema", t.schema, "cols", [c.column_name for c in t.columns])