From 27d31ba1e2e974594fa5ea9961956336c90d1784 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 9 Jul 2026 23:56:30 -0700 Subject: [PATCH 1/7] [adapters] Increase wait_for_records timeout in tests. suspend_barrier5 times out waiting for 4000 input records while suspending in CI. The log seems to indicate that the connector is making progress, just slowly under load. We increase the timeout from 10 to 100s and improve progress logging. If the issue occurs again, this means there's a real bug somewhere. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/controller/test.rs | 81 +++++++++++++++++++------- 1 file changed, 59 insertions(+), 22 deletions(-) diff --git a/crates/adapters/src/controller/test.rs b/crates/adapters/src/controller/test.rs index 6f2e34368a9..f19f98f3cde 100644 --- a/crates/adapters/src/controller/test.rs +++ b/crates/adapters/src/controller/test.rs @@ -431,9 +431,40 @@ fn collect_endpoint_records(controller: &Controller, n: usize) -> Vec { .collect() } +/// Default budget for [`wait_for_records`] to observe the expected output. +const OUTPUT_TIMEOUT_MS: u128 = 10_000; + +/// Longer output budget for waits taken while a suspend is in progress. +/// +/// While a suspend is pending, the controller advances only the barrier inputs +/// while re-attempting the checkpoint each step, so under CI CPU contention +/// output can lag far behind input even though stepping keeps making forward +/// progress. Using the default [`OUTPUT_TIMEOUT_MS`] in that window makes the +/// suspend tests flake. Matches the 100s budget already used for the +/// suspend-completion `recv_timeout` in these tests. +const SUSPEND_OUTPUT_TIMEOUT_MS: u128 = 100_000; + +/// Like `println!`, but prefixes a UTC timestamp so test output interleaves +/// legibly with the controller's tracing logs when diagnosing timing issues. +macro_rules! tprintln { + ($($arg:tt)*) => { + println!("[{}] {}", chrono::Utc::now().format("%H:%M:%S%.6f"), format_args!($($arg)*)) + }; +} + +/// Wait until every output endpoint has received at least the expected number +/// of records, then verify the counts match exactly, using the default +/// [`OUTPUT_TIMEOUT_MS`] budget. #[track_caller] fn wait_for_records(controller: &Controller, expect_n: &[usize]) { - println!("waiting for {expect_n:?} records..."); + wait_for_records_within(controller, expect_n, OUTPUT_TIMEOUT_MS); +} + +/// Like [`wait_for_records`], but waits up to `timeout_ms` for the records to +/// arrive. +#[track_caller] +fn wait_for_records_within(controller: &Controller, expect_n: &[usize], timeout_ms: u128) { + tprintln!("waiting for {expect_n:?} records..."); let n = expect_n.len(); let mut last_n = repeat_n(0, n).collect::>(); wait( @@ -441,7 +472,7 @@ fn wait_for_records(controller: &Controller, expect_n: &[usize]) { let new_n = collect_endpoint_records(controller, n); for i in 0..n { if new_n[i] > last_n[i] { - println!("received {} records on test_output{}", new_n[i], i + 1); + tprintln!("received {} records on test_output{}", new_n[i], i + 1); } } last_n = new_n; @@ -450,7 +481,7 @@ fn wait_for_records(controller: &Controller, expect_n: &[usize]) { .zip(expect_n.iter()) .all(|(&last, &expect)| last >= expect) }, - 10_000, + timeout_ms, ) .unwrap(); @@ -2028,25 +2059,25 @@ fn suspend_barrier() { .from_writer(&input_file); // Write records to the input file. - println!("Writing records 0..4000"); + tprintln!("Writing records 0..4000"); for id in 0..4000 { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); } writer.flush().unwrap(); // Start pipeline. - println!("start pipeline"); + tprintln!("start pipeline"); let controller = start_controller(&storage_dir, &[5000]); // Wait for the records that are not in the checkpoint to be // processed or replayed. - println!("wait for 4000 records 0..4000"); + tprintln!("wait for 4000 records 0..4000"); wait_for_records(&controller, &[4000]); // Suspend. let (sender, receiver) = mpsc::channel(); - println!("start suspend"); + tprintln!("start suspend"); let suspend_request_step = controller.status().global_metrics.total_initiated_steps(); controller.start_suspend(Box::new(move |result| sender.send(result).unwrap())); @@ -2068,7 +2099,7 @@ fn suspend_barrier() { assert_bounded_suspend_steps(&controller, suspend_request_step); // Stop controller. - println!("stop controller"); + tprintln!("stop controller"); controller.stop().unwrap(); // Read output and compare. Our output adapter, which is not @@ -2082,7 +2113,7 @@ fn suspend_barrier() { let controller = start_controller(&storage_dir, &[5000]); wait_for_records(&controller, &[5000]); - println!("start suspend"); + tprintln!("start suspend"); let (sender, receiver) = mpsc::channel(); let mut sender = Some(sender); @@ -2092,13 +2123,16 @@ fn suspend_barrier() { let start = (i + 5) * 1000; let end = start + 1000; - println!("writing records {start}..{end}"); + tprintln!("writing records {start}..{end}"); for id in start..end { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); } writer.flush().unwrap(); - println!("waiting for {end} records"); - wait_for_records(&controller, &[end]); + tprintln!("waiting for {end} records"); + // After the first iteration a suspend is pending: the controller only + // advances barrier inputs while re-attempting the checkpoint each step, + // so output can lag under CI CPU load. Use the longer budget. + wait_for_records_within(&controller, &[end], SUSPEND_OUTPUT_TIMEOUT_MS); check_file_contents(&(output_path(&storage_dir, 0)), 5000..end); if let Some(sender) = sender.take() { @@ -2157,7 +2191,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { .collect::>(); // Write records to the input files. - println!("Writing 1000 records to each of {n_inputs} files"); + tprintln!("Writing 1000 records to each of {n_inputs} files"); for writer in writers.iter_mut().take(n_inputs) { for id in 0..1000 { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); @@ -2166,7 +2200,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { } // Start pipeline. - println!("start pipeline"); + tprintln!("start pipeline"); // The barrier for input 0 is record 0, // for input 1 is record 1000, @@ -2177,7 +2211,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // Wait for the first 1000 records in each file to be read and copied to the // output. - println!("wait for 1000 records in each file"); + tprintln!("wait for 1000 records in each file"); let mut written = repeat_n(1000, n_inputs).collect::>(); wait_for_records(&controller, &written); @@ -2187,7 +2221,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // we're past all the barriers; otherwise, it will not complete due to // barriers, since each input only has 1000 records so far. let (sender, receiver) = mpsc::channel(); - println!("start suspend"); + tprintln!("start suspend"); let suspend_request_step = controller.status().global_metrics.total_initiated_steps(); controller.start_suspend(Box::new(move |result| sender.send(result).unwrap())); @@ -2212,7 +2246,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { receiver.try_recv().unwrap_err(); // Write 1000 more records to the `next` input. - println!("writing 1000 more records to test_input{}", next + 1); + tprintln!("writing 1000 more records to test_input{}", next + 1); for id in written[next]..written[next] + 1000 { writers[next] .serialize(TestStruct::for_id(id as u32)) @@ -2228,9 +2262,12 @@ fn suspend_multiple_barriers(n_inputs: usize) { // We won't get any more records on output from inputs that have reached // their barrier, so the writes to inputs 0 and 1 won't have any effect // here. - println!("total written: {written:?}"); + tprintln!("total written: {written:?}"); let expect = expectations(&written, &barriers); - wait_for_records(&controller, &expect); + // A suspend is pending here: the controller only advances barrier inputs + // while re-attempting the checkpoint each step, so output can lag far + // behind under CI CPU load. Use the longer budget. + wait_for_records_within(&controller, &expect, SUSPEND_OUTPUT_TIMEOUT_MS); for (i, expectation) in expect.iter().enumerate().take(n_inputs) { check_file_contents(&output_path(&storage_dir, i), 0..*expectation); } @@ -2244,11 +2281,11 @@ fn suspend_multiple_barriers(n_inputs: usize) { assert_bounded_suspend_steps(&controller, suspend_request_step); // Stop controller. - println!("stop controller"); + tprintln!("stop controller"); controller.stop().unwrap(); // Check output one more time. - println!("check output one more time now that controller is stopped"); + tprintln!("check output one more time now that controller is stopped"); let expect = expectations(&written, &barriers); for (i, e) in expect.iter().enumerate().take(n_inputs) { check_file_contents(&output_path(&storage_dir, i), 0..*e); @@ -2256,7 +2293,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // Now restart the controller and wait for all the records that we wrote // beyond the barriers get copied to output (and nothing else). - println!("restart controller and wait for records beyond the barriers"); + tprintln!("restart controller and wait for records beyond the barriers"); let controller = start_controller(&storage_dir, &barriers); wait_for_records(&controller, &written); for i in 0..n_inputs { From b203ac33e2826a0630f40f726a6f97605a0cc8e6 Mon Sep 17 00:00:00 2001 From: Anand Raman Date: Tue, 7 Jul 2026 17:29:22 -0500 Subject: [PATCH 2/7] Lakeflow Declarative Pipelines and Feldera Demo Added configuration for a Lakeflow declarative pipeline. Users can copy the bronze data to their own bucket, see how Feldera works with snapshot_and_follow mode, and contrast with Databricks IVM solution (lakeflow declarative pipelines) Signed-off-by: Anand Raman --- docs.feldera.com/docs/sidebars.js | 3 +- .../databricks/ingest_bronze.py | 311 +++++++++++ .../databricks/job.yaml | 37 ++ .../databricks/measure_latency.py | 87 +++ .../databricks/pipeline.yaml | 16 + .../databricks/silver_gold_pipeline.py | 497 ++++++++++++++++++ .../use_cases/medallion_architecture/part3.md | 2 + .../use_cases/medallion_architecture/part4.md | 305 +++++++++++ 8 files changed, 1257 insertions(+), 1 deletion(-) create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/ingest_bronze.py create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/measure_latency.py create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/silver_gold_pipeline.py create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/part4.md diff --git a/docs.feldera.com/docs/sidebars.js b/docs.feldera.com/docs/sidebars.js index f7342618e17..fbbbe29773e 100644 --- a/docs.feldera.com/docs/sidebars.js +++ b/docs.feldera.com/docs/sidebars.js @@ -120,7 +120,8 @@ const guides = { items: [ 'use_cases/medallion_architecture/part1', 'use_cases/medallion_architecture/part2', - 'use_cases/medallion_architecture/part3' + 'use_cases/medallion_architecture/part3', + 'use_cases/medallion_architecture/part4' ] }, { diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/ingest_bronze.py b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/ingest_bronze.py new file mode 100644 index 00000000000..e7156f89613 --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/ingest_bronze.py @@ -0,0 +1,311 @@ +# Databricks notebook source +# MAGIC %md +# MAGIC # Step 1 — Ingest CDC into the Bronze Delta tables +# MAGIC +# MAGIC First task of the `compute_silver_gold` Databricks Workflow. It does two things: +# MAGIC +# MAGIC 1. **(optional) `clean_start`** — when the job parameter `clean_start = true`, the +# MAGIC working bronze Delta tables are deleted and re-seeded from the read-only source +# MAGIC snapshot at `s3://feldera-demos/ecommerce-cdc-{scale}/snapshot`. The silver and +# MAGIC gold materialized views are then recomputed by the pipeline refresh step — a +# MAGIC normal refresh, which Enzyme automatically turns into a full recompute once it +# MAGIC sees bronze was rewritten. +# MAGIC 2. **CDC ingest** — read the Feldera-format CDC NDJSON file for the single hour +# MAGIC `hour` and apply it to the working bronze Delta tables on S3. `bronze_orders` +# MAGIC carries status updates (delete-of-old + insert-of-new), so it is MERGEd by +# MAGIC `order_id` keeping the latest row; the other three fact tables are append-only +# MAGIC event streams, so their inserts are simply appended. +# MAGIC +# MAGIC Run the hours in order (`...T00`, `...T01`, ...): each run applies exactly one +# MAGIC hour of change, the same granularity at which `push_changes.py --hour` feeds +# MAGIC Feldera. +# MAGIC +# MAGIC After this notebook finishes, the workflow refreshes the silver/gold materialized +# MAGIC views defined in `silver_gold_pipeline.py`. +# MAGIC +# MAGIC **Feldera equivalent:** there is no separate ingest step — Feldera follows the +# MAGIC bronze Delta tables and propagates each change through the view DAG incrementally. + +# COMMAND ---------- + +from pyspark.sql import functions as F +from pyspark.sql.window import Window + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Parameters + +# COMMAND ---------- + +# CDC spans 7 days x 24 hours (generate_snapshot_cdc.py): 2025-11-30T00 .. 2025-12-06T23. +# Offer those as a picklist, plus "" for a seed-only run (no CDC). +from datetime import datetime, timedelta + +CDC_FIRST_DAY = datetime(2025, 11, 30) +CDC_NUM_DAYS = 7 +HOUR_CHOICES = [""] + [ + (CDC_FIRST_DAY + timedelta(hours=h)).strftime("%Y-%m-%dT%H") + for h in range(CDC_NUM_DAYS * 24) +] + +dbutils.widgets.dropdown("hour", "", HOUR_CHOICES, "hour (empty = seed only)") +dbutils.widgets.text("scale_factor", "0.01", "scale_factor") +dbutils.widgets.dropdown("clean_start", "false", ["false", "true"], "clean_start") +dbutils.widgets.text("source_bucket", "s3://feldera-demos", "source_bucket") +dbutils.widgets.text("warehouse_bucket", "s3://feldera-demos", "warehouse_bucket") + +HOUR = dbutils.widgets.get("hour").strip() or None +SCALE_FACTOR = float(dbutils.widgets.get("scale_factor")) +CLEAN_START = dbutils.widgets.get("clean_start").strip().lower() == "true" +SOURCE_BUCKET = dbutils.widgets.get("source_bucket").strip() +WAREHOUSE_BUCKET = dbutils.widgets.get("warehouse_bucket").strip() + +SCALE_STR = str(SCALE_FACTOR).replace(".", "-") + +# Read-only source: the snapshot + CDC produced by generate_snapshot_cdc.py. +# SOURCE_BUCKET comes from the source_bucket job parameter (read above). +SOURCE_PREFIX = f"ecommerce-cdc-{SCALE_STR}" +SNAPSHOT_ROOT = f"{SOURCE_BUCKET}/{SOURCE_PREFIX}/snapshot" +CDC_ROOT = f"{SOURCE_BUCKET}/{SOURCE_PREFIX}/cdc" + +# Writable working warehouse: the bronze tables this workflow ingests into and the +# silver/gold materialized-view pipeline reads from. Kept separate from the read-only +# source so a clean_start can rebuild it from scratch without touching the source. +# WAREHOUSE_BUCKET comes from the warehouse_bucket job parameter (read above), the same +# value the pipeline reads via spark.conf — single source of truth across both tasks. +WAREHOUSE_PREFIX = f"ecommerce-pipeline-{SCALE_STR}" +BRONZE_ROOT = f"{WAREHOUSE_BUCKET}/{WAREHOUSE_PREFIX}/bronze" + + +def s3a(path: str) -> str: + return path.replace("s3://", "s3a://") + + +print(f"clean_start: {CLEAN_START}") +print(f"hour: {HOUR or '(none — seed only)'}") +print(f"scale_factor: {SCALE_FACTOR}") +print(f"source: {SNAPSHOT_ROOT}") +print(f"cdc: {CDC_ROOT}") +print(f"bronze (rw): {BRONZE_ROOT}") + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Table metadata +# MAGIC +# MAGIC All seven bronze tables are re-seeded on `clean_start`. Only the four fact tables +# MAGIC receive CDC, and they split into two ingest modes: +# MAGIC +# MAGIC - **`merge`** — `bronze_orders` only. Orders go pending -> confirmed -> shipped -> +# MAGIC delivered; each transition is a delete-of-old + insert-of-new, so the hour's +# MAGIC inserts are deduped to the latest `updated_at` per `order_id` and MERGEd. +# MAGIC - **`append`** — order items, clickstream and inventory are append-only event +# MAGIC streams that never update an existing row, so the hour's inserts are appended +# MAGIC directly (no join against the target, no file rewrites). + +# COMMAND ---------- + +# All bronze tables. Used for the clean_start re-seed (DEEP CLONE from snapshot). +BRONZE_TABLES = [ + "bronze_suppliers", + "bronze_products", + "bronze_customers", + "bronze_orders", + "bronze_order_items", + "bronze_clickstream_events", + "bronze_inventory_events", +] + +# Fact tables that stream CDC -> how to apply one hour of change. +# merge: upsert keyed on `pk`, keeping the latest `order_by` row per key. +# append: insert-only; append the hour's new rows. +CDC_TABLES = { + "bronze_orders": { + "short": "orders", + "mode": "merge", + "pk": "order_id", + "order_by": "updated_at", + }, + "bronze_order_items": {"short": "order_items", "mode": "append"}, + "bronze_clickstream_events": {"short": "clickstream_events", "mode": "append"}, + "bronze_inventory_events": {"short": "inventory_events", "mode": "append"}, +} + + +def bronze_path(name: str) -> str: + return s3a(f"{BRONZE_ROOT}/{name}") + + +def snapshot_path(name: str) -> str: + return s3a(f"{SNAPSHOT_ROOT}/{name}") + + +def bronze_exists(name: str) -> bool: + """True if a Delta table is present at the working bronze path for `name` + (checks for its _delta_log). Uses the s3:// form for dbutils.fs.""" + try: + dbutils.fs.ls(f"{BRONZE_ROOT}/{name}/_delta_log") + return True + except Exception: # noqa: BLE001 — path absent => table not seeded + return False + + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## clean_start — delete and re-seed bronze from the source snapshot +# MAGIC +# MAGIC Deletes the working bronze path, then copies each bronze table fresh from the +# MAGIC read-only source snapshot via DEEP CLONE. Silver/gold are materialized views owned +# MAGIC by the Lakeflow pipeline (not stored at these warehouse paths), so they aren't +# MAGIC touched here — they recompute on the pipeline refresh step, which Enzyme turns into +# MAGIC a full recompute once it sees bronze was rewritten. + +# COMMAND ---------- + +if CLEAN_START: + print("clean_start = true -> deleting bronze re-seed") + layer_path = f"{WAREHOUSE_BUCKET}/{WAREHOUSE_PREFIX}/bronze" + try: + dbutils.fs.rm(layer_path, recurse=True) + print(f" deleted {layer_path}") + except Exception as e: # noqa: BLE001 — best-effort delete of a possibly-absent path + print(f" skip {layer_path}: {e}") + + for name in BRONZE_TABLES: + src = snapshot_path(name) + dst = bronze_path(name) + # DEEP CLONE copies the source data files directly (and their Delta stats) + # instead of reading every row back through Spark and re-encoding it on write. + # The row count comes from the operation metrics, so there is no extra scan. + # Enable row tracking on the working bronze: the silver/gold pipeline's incremental + # materialized views require row tracking on their source tables, else the refresh + # fails with ROW_TRACKING_NOT_ENABLED. Set on create so the later MERGE/append + # writes preserve it. + res = spark.sql( + f"CREATE OR REPLACE TABLE delta.`{dst}` " + f"DEEP CLONE delta.`{src}` " + "TBLPROPERTIES ('delta.enableRowTracking' = 'true')" + ).collect() + copied = res[0]["num_copied_files"] if res else 0 + print(f" re-seeded {name}: cloned {copied} data files") + print( + "clean_start complete — bronze tables are a fresh copy of the source snapshot" + ) +else: + print("clean_start = false -> keeping existing bronze tables, applying CDC on top") + # clean_start=false means we apply CDC on top of existing bronze, so both an hour to + # ingest and pre-seeded bronze tables are required. Fail fast with a clear message + # rather than no-op'ing here and erroring later in the pipeline's bronze reads. + if HOUR is None: + raise ValueError( + "hour is required when clean_start=false (nothing to ingest otherwise). " + "Pass hour=YYYY-MM-DDThh, or set clean_start=true to seed bronze first." + ) + missing = [t for t in BRONZE_TABLES if not bronze_exists(t)] + if missing: + raise FileNotFoundError( + f"Bronze tables missing under {BRONZE_ROOT}: {missing}. " + "Run once with clean_start=true to seed bronze from the source snapshot " + "before applying CDC." + ) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## CDC ingest — apply one hour of Feldera-format change records to bronze +# MAGIC +# MAGIC Each CDC line is `{"insert": {...}}` / `{"delete": {...}}` +# MAGIC ([Feldera JSON format](https://docs.feldera.com/formats/json)); a status update is +# MAGIC a delete of the old row followed by an insert of the new one. We keep the inserted +# MAGIC state (the new/updated row) and drop the deleted state — an update re-inserts the +# MAGIC new row, which wins via the orders MERGE key, so the inserts alone reconstruct the +# MAGIC current table. Only one hour's file (`{hour}.json`) is read per table. + +# COMMAND ---------- + + +def read_hour_inserts(table_short_name, target_schema, hour): + """Read the single CDC file for `hour` and return its inserted row state, cast to + the target (bronze) schema. Returns None if that hour has no file for this table.""" + path = s3a(f"{CDC_ROOT}/{table_short_name}/{hour}.json") + try: + # spark.read.json on a missing path throws; a present-but-empty file yields no + # rows. Either way we treat it as "no change this hour". + raw = spark.read.json(path) + except Exception: # noqa: BLE001 — no file for this hour/table + return None + if "insert" not in raw.columns: + return None + + records = raw.filter(F.col("insert").isNotNull()).select("insert.*") + if not records.head(1): + return None + + # JSON serializes decimals/timestamps as strings — cast back to the bronze schema. + for field in target_schema.fields: + if field.name in records.columns: + records = records.withColumn( + field.name, F.col(field.name).cast(field.dataType) + ) + + return records.select([f.name for f in target_schema.fields]) + + +def ingest_hour(table_name, hour): + """Apply one hour of CDC to `table_name`: MERGE for orders, append otherwise.""" + cfg = CDC_TABLES[table_name] + dst = bronze_path(table_name) + target_schema = spark.read.format("delta").load(dst).schema + + cdc = read_hour_inserts(cfg["short"], target_schema, hour) + if cdc is None: + print(f" {table_name}: no CDC for {hour} — unchanged") + return + + if cfg["mode"] == "append": + # Insert-only stream: append the hour's rows. No join against the target and no + # file rewrites — the cheapest possible Delta write. + cdc.write.format("delta").mode("append").save(dst) + print(f" {table_name}: appended {cdc.count():,} rows") + return + + # merge mode (orders): the hour may carry several transitions for one order_id; + # keep the latest updated_at, then upsert. MERGE returns a metrics row, so the + # counts come for free without a second pass over the source DAG. + pk, order_by = cfg["pk"], cfg["order_by"] + w = Window.partitionBy(pk).orderBy(F.col(order_by).desc()) + latest = ( + cdc.withColumn("_rn", F.row_number().over(w)) + .filter(F.col("_rn") == 1) + .drop("_rn") + ) + + latest.createOrReplaceTempView("cdc_src") + metrics = ( + spark.sql(f""" + MERGE INTO delta.`{dst}` AS t + USING cdc_src AS s + ON t.{pk} = s.{pk} + WHEN MATCHED THEN UPDATE SET * + WHEN NOT MATCHED THEN INSERT * + """) + .collect()[0] + .asDict() + ) + print( + f" {table_name}: {metrics.get('num_inserted_rows', 0):,} inserted, " + f"{metrics.get('num_updated_rows', 0):,} updated (keyed on {pk})" + ) + + +# COMMAND ---------- + +if HOUR is None: + print("hour is empty — skipping CDC ingest (bronze left at its seeded state).") +else: + print(f"Applying CDC hour {HOUR} to bronze...") + for table_name in CDC_TABLES: + ingest_hour(table_name, HOUR) + print("Bronze ingest complete. Silver/gold materialized views refresh next.") diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml new file mode 100644 index 00000000000..2d59db54473 --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml @@ -0,0 +1,37 @@ +# Job settings for the compute_silver_gold workflow. +# Create via the Jobs UI (Create job -> kebab menu -> Edit as YAML) or the jobs API. +# - Replace the ingest_bronze notebook_path with where you imported ingest_bronze.py. +# - Replace pipeline_id with the id of the pipeline created from pipeline.yaml. +# - Assign serverless (or other) compute to the notebook task in the UI. +# scale_factor + warehouse_bucket are passed to the ingest notebook here and also override +# the pipeline's parameters of the same key when the pipeline task runs. +resources: + jobs: + name: compute_silver_gold + parameters: + - name: hour + default: "" + - name: clean_start + default: "false" + - name: scale_factor + default: "0.01" + - name: source_bucket + default: s3://feldera-demos + - name: warehouse_bucket + default: + tasks: + - task_key: ingest_bronze + notebook_task: + notebook_path: + base_parameters: + hour: "{{job.parameters.hour}}" + clean_start: "{{job.parameters.clean_start}}" + scale_factor: "{{job.parameters.scale_factor}}" + source_bucket: "{{job.parameters.source_bucket}}" + warehouse_bucket: "{{job.parameters.warehouse_bucket}}" + - task_key: refresh_silver_gold + depends_on: + - task_key: ingest_bronze + pipeline_task: + pipeline_id: "" + full_refresh: false diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/measure_latency.py b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/measure_latency.py new file mode 100644 index 00000000000..c5b4a0bf37e --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/measure_latency.py @@ -0,0 +1,87 @@ +# Measures end-to-end latency for the Delta Lake input connectors using the +# completed_frontier timestamps exposed on /stats. +# See https://docs.feldera.com/pipelines/latency/#measuring-end-to-end-latency-with-input-frontiers +# +# For each Delta connector the frontier reports three RFC 3339 timestamps: +# ingested_at -- data for this frontier arrived at the connector +# processed_at -- the IVM engine finished processing it +# completed_at -- outputs reached all sinks +# Latency is the delta between these, plus freshness = now - completed_at. +import os +from argparse import ArgumentParser +from datetime import datetime, timezone + +from dotenv import load_dotenv +from feldera import FelderaClient + +PIPELINE = "ecommerce-medallion-architecture" + +# Bronze tables to report on (user-facing names map to bronze_). +TABLES = [ + "bronze_clickstream_events", + "bronze_inventory_events", + "bronze_orders", + "bronze_order_items", +] + + +def parse_ts(value: str) -> datetime: + # Frontier timestamps are RFC 3339 with a trailing Z; normalize for fromisoformat. + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def delta_connector(inputs: list[dict], table: str) -> dict | None: + # A Delta table has both an api-ingress and a delta_table_input connector on + # the same stream; only the Delta one populates completed_frontier. + for i in inputs: + if i["endpoint_name"].startswith(f"{table}.") and i.get("completed_frontier"): + return i + return None + + +def report(client: FelderaClient) -> None: + stats = client.get_pipeline_stats(PIPELINE) + now = datetime.now(timezone.utc) + + header = ( + f"{'table':<26}{'version':>8}{'ingest→proc':>14}{'proc→done':>12}{'e2e':>10}" + ) + print(header) + print("-" * len(header)) + + for table in TABLES: + conn = delta_connector(stats["inputs"], table) + if conn is None: + print(f"{table:<26}{'no delta frontier yet':>56}") + continue + + f = conn["completed_frontier"] + version = f.get("metadata", {}).get("version") + ingested = parse_ts(f["ingested_at"]) + processed = parse_ts(f["processed_at"]) + completed = parse_ts(f["completed_at"]) + + ingest_to_proc = (processed - ingested).total_seconds() + proc_to_done = (completed - processed).total_seconds() + e2e = (completed - ingested).total_seconds() + + print( + f"{table:<26}{version:>8}" + f"{ingest_to_proc:>13.2f}s{proc_to_done:>11.2f}s" + f"{e2e:>9.2f}s" + ) + + +if __name__ == "__main__": + parser = ArgumentParser( + description="Measure Delta input latency via /stats frontiers" + ) + parser.add_argument("--pipeline", default=PIPELINE) + args = parser.parse_args() + PIPELINE = args.pipeline + + load_dotenv() + client = FelderaClient( + url=os.getenv("FELDERA_URL"), api_key=os.getenv("FELDERA_API_KEY") + ) + report(client) diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml new file mode 100644 index 00000000000..4b8e08f9c50 --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml @@ -0,0 +1,16 @@ +# Lakeflow Declarative Pipeline settings (the "transformation" — runs silver_gold_pipeline.py). +# Create via the Pipelines UI (Create pipeline -> Settings -> YAML) or the pipelines API. +# - Replace the notebook path with where you imported silver_gold_pipeline.py. +# - Set catalog/schema to where the silver/gold materialized views should publish. +# configuration values are read via spark.conf in the notebook; the compute_silver_gold job +# passes job parameters of the same key, which override these at run time. +name: ecommerce-silver-gold-mv +serverless: true +catalog: +schema: +configuration: + scale_factor: "0.01" + warehouse_bucket: +libraries: + - notebook: + path: diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/silver_gold_pipeline.py b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/silver_gold_pipeline.py new file mode 100644 index 00000000000..632e1accaf0 --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/silver_gold_pipeline.py @@ -0,0 +1,497 @@ +# Databricks notebook source +# MAGIC %md +# MAGIC # Silver & Gold — Databricks Materialized Views (Lakeflow Declarative Pipeline) +# MAGIC +# MAGIC The silver and gold layers from `compute_silver_gold.py`, expressed as a Lakeflow +# MAGIC Declarative Pipeline. Each silver/gold dataset is a **materialized view** that +# MAGIC Databricks refreshes incrementally where the query allows (Enzyme), falling back +# MAGIC to a full recompute otherwise. +# MAGIC +# MAGIC - **Bronze** is exposed as lightweight `@dp.temporary_view`s that read the working +# MAGIC bronze Delta tables on S3 (written by `ingest_bronze.py`). They are not persisted — +# MAGIC the bronze tables already live in S3. +# MAGIC - **Silver / Gold** are `@dp.materialized_view`s, persisted in the pipeline's Unity +# MAGIC Catalog target schema. +# MAGIC +# MAGIC The business logic (SQL) is identical to the batch script and to the Feldera views +# MAGIC in `ecommerce/sql/ecommerce.sql`. The workflow refreshes this pipeline as its second +# MAGIC step with a single normal refresh: it updates incrementally after a CDC hour, and +# MAGIC Enzyme automatically full-recomputes after a `clean_start` rewrites bronze. +# MAGIC +# MAGIC **Feldera equivalent:** the same DAG, maintained incrementally on every commit with +# MAGIC no pipeline scheduling and no full-refresh fallback. + +# COMMAND ---------- + +from pyspark import pipelines as dp +from pyspark.sql import functions as F + +# scale_factor is supplied as pipeline configuration; it selects the bronze S3 path the +# ingest step wrote to. Defaults to 0.01 to match the rest of the demo. +SCALE_FACTOR = float(spark.conf.get("scale_factor", "0.01")) +SCALE_STR = str(SCALE_FACTOR).replace(".", "-") + +WAREHOUSE_BUCKET = spark.conf.get("warehouse_bucket") +BRONZE_ROOT = f"{WAREHOUSE_BUCKET}/ecommerce-pipeline-{SCALE_STR}/bronze".replace( + "s3://", "s3a://" +) + +# The silver/gold materialized views publish to the pipeline's configured catalog/schema +# (set on the pipeline when it's created — see pipeline.yaml catalog/schema). Datasets use +# bare names and resolve within that destination. + + +def bronze_path(name: str) -> str: + return f"{BRONZE_ROOT}/{name}" + + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Bronze source views +# MAGIC +# MAGIC Pass-through views over the working bronze Delta tables on S3. Declaring them as +# MAGIC pipeline datasets lets the silver/gold materialized views reference them by name +# MAGIC (in `spark.sql`) and lets Enzyme reason about their changes. + +# COMMAND ---------- + +_BRONZE = [ + "bronze_suppliers", + "bronze_products", + "bronze_customers", + "bronze_orders", + "bronze_order_items", + "bronze_clickstream_events", + "bronze_inventory_events", +] + + +def _make_bronze_view(table_name): + @dp.temporary_view(name=table_name) + def _view(): + return spark.read.format("delta").load(bronze_path(table_name)) + + return _view + + +for _name in _BRONZE: + _make_bronze_view(_name) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Silver layer +# MAGIC +# MAGIC Each function runs the exact SQL from `compute_silver_gold.py`, referencing its +# MAGIC upstream pipeline datasets directly by name — Lakeflow resolves those names against +# MAGIC the pipeline graph and registers them as dependencies, so the logic stays in +# MAGIC lock-step with the batch script and the Feldera views. + +# COMMAND ---------- + + +@dp.materialized_view( + name="silver_customers", + comment="Cleaned customer dimension (valid tiers).", + refresh_policy="incremental", +) +def silver_customers(): + return spark.read.table("bronze_customers").filter( + F.col("customer_id").isNotNull() + & F.col("customer_tier").isin("standard", "silver", "gold", "platinum") + ) + + +@dp.materialized_view( + name="silver_products", + comment="Cleaned product dimension (product grain).", + refresh_policy="incremental", +) +def silver_products(): + return spark.sql(""" + SELECT product_id, product_name, category, brand, list_price + FROM bronze_products + WHERE product_id IS NOT NULL + """) + + +@dp.materialized_view( + name="silver_enriched_clickstream", + comment="Clickstream filtered to interaction events (clean-only).", + refresh_policy="incremental", +) +def silver_enriched_clickstream(): + return spark.sql(""" + SELECT + ce.event_id, ce.user_id, ce.session_id, ce.event_type, ce.page_url, + ce.product_id, ce.device_type, ce.geo_country, ce.geo_region, + ce.event_timestamp + FROM bronze_clickstream_events ce + WHERE ce.event_type IS NOT NULL + AND ce.user_id IS NOT NULL + AND ce.event_type IN ('page_view', 'product_view', 'add_to_cart', 'begin_checkout', 'purchase') + """) + + +@dp.materialized_view( + name="silver_orders_enriched", + comment="Orders joined to customers + aggregated order items.", + refresh_policy="incremental", +) +def silver_orders_enriched(): + return spark.sql(""" + SELECT + o.order_id, o.user_id, o.order_status, o.order_total, + o.discount_amount, o.shipping_cost, o.payment_method, o.coupon_code, + o.created_at, o.updated_at, + c.customer_tier, + c.geo_country AS customer_country, + c.geo_region AS customer_region, + c.signup_date, + oi.item_count, oi.total_quantity, oi.gross_item_revenue, oi.avg_discount_pct + FROM bronze_orders o + JOIN silver_customers c ON o.user_id = c.customer_id + JOIN ( + SELECT order_id, + COUNT(*) AS item_count, + SUM(quantity) AS total_quantity, + SUM(quantity * unit_price) AS gross_item_revenue, + AVG(discount_pct) AS avg_discount_pct + FROM bronze_order_items + WHERE quantity > 0 AND unit_price >= 0 + GROUP BY order_id + ) oi ON o.order_id = oi.order_id + WHERE o.order_id IS NOT NULL + AND o.user_id IS NOT NULL + AND o.order_total >= 0 + AND o.order_status IN ('pending', 'confirmed', 'shipped', 'delivered', 'cancelled', 'returned') + """) + + +@dp.materialized_view( + name="silver_order_items_enriched", + comment="Order items joined to products, suppliers, orders, customers with margins.", + refresh_policy="incremental", +) +def silver_order_items_enriched(): + return spark.sql(""" + SELECT + oi.order_item_id, oi.order_id, oi.product_id, + oi.quantity, oi.unit_price, oi.discount_pct, + oi.quantity * oi.unit_price AS line_gross_revenue, + oi.quantity * oi.unit_price * (1.0 - COALESCE(oi.discount_pct, 0) / 100.0) AS line_net_revenue, + oi.quantity * (oi.unit_price - p.cost_price) AS line_gross_margin, + p.product_name, p.category, p.brand, p.cost_price, p.list_price, + s.supplier_id, s.supplier_name, s.country AS supplier_country, s.lead_time_days, + o.created_at AS order_created_at, + o.order_status, o.user_id, + c.customer_tier + FROM bronze_order_items oi + JOIN bronze_products p ON oi.product_id = p.product_id + JOIN bronze_suppliers s ON p.supplier_id = s.supplier_id + JOIN bronze_orders o ON oi.order_id = o.order_id + JOIN silver_customers c ON o.user_id = c.customer_id + WHERE oi.quantity > 0 + AND oi.unit_price >= 0 + AND p.is_active = TRUE + AND o.order_status IN ('pending', 'confirmed', 'shipped', 'delivered', 'cancelled', 'returned') + """) + + +@dp.materialized_view( + name="silver_confirmed_order_items", + comment="Order items excluding cancelled/returned orders.", + refresh_policy="incremental", +) +def silver_confirmed_order_items(): + return spark.sql(""" + SELECT * FROM silver_order_items_enriched + WHERE order_status NOT IN ('cancelled', 'returned') + """) + + +@dp.materialized_view( + name="silver_inventory_current", + comment="Running stock per product/warehouse from inventory events.", + refresh_policy="incremental", +) +def silver_inventory_current(): + return spark.sql(""" + SELECT + ie.product_id, ie.warehouse_id, + p.product_name, p.category, p.brand, + s.supplier_id, s.supplier_name, s.lead_time_days, + SUM(ie.quantity_change) AS current_stock, + SUM(CASE WHEN ie.event_type = 'restock' THEN ie.quantity_change ELSE 0 END) AS total_restocked, + SUM(CASE WHEN ie.event_type = 'sale_reserve' THEN ABS(ie.quantity_change) ELSE 0 END) + - SUM(CASE WHEN ie.event_type = 'cancellation_restock' THEN ie.quantity_change ELSE 0 END) + AS total_sold, + SUM(CASE WHEN ie.event_type = 'return_restock' THEN ie.quantity_change ELSE 0 END) AS total_returned + FROM bronze_inventory_events ie + JOIN bronze_products p ON ie.product_id = p.product_id + JOIN bronze_suppliers s ON p.supplier_id = s.supplier_id + WHERE ie.quantity_change IS NOT NULL AND ie.product_id IS NOT NULL + GROUP BY ie.product_id, ie.warehouse_id, p.product_name, p.category, p.brand, s.supplier_id, s.supplier_name, s.lead_time_days + """) + + +@dp.materialized_view( + name="silver_inventory_by_supplier", + comment="Stock rolled up to supplier grain.", + refresh_policy="incremental", +) +def silver_inventory_by_supplier(): + return spark.sql(""" + SELECT supplier_id, supplier_name, + SUM(current_stock) AS total_current_stock, + SUM(total_sold) AS total_sold, + SUM(total_restocked) AS total_restocked, + SUM(total_returned) AS total_returned + FROM silver_inventory_current + GROUP BY supplier_id, supplier_name + """) + + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Gold layer + +# COMMAND ---------- + + +@dp.materialized_view( + name="gold_order_status_summary", + comment="Order status distribution.", + refresh_policy="incremental", +) +def gold_order_status_summary(): + return spark.sql(""" + SELECT + o.order_status, + COUNT(DISTINCT o.order_id) AS order_count, + COALESCE(SUM(o.order_total), 0) AS total_revenue, + AVG(o.order_total) AS avg_order_value, + COUNT(DISTINCT o.user_id) AS unique_customers + FROM silver_orders_enriched o + GROUP BY o.order_status + """) + + +@dp.materialized_view( + name="gold_supplier_performance", + comment="Revenue + margin per supplier.", + refresh_policy="incremental", +) +def gold_supplier_performance(): + return spark.sql(""" + WITH orders_by_supplier AS ( + SELECT + oi.supplier_id, oi.supplier_name, oi.supplier_country, oi.lead_time_days, + SUM(oi.quantity) AS total_units_sold, + SUM(oi.line_net_revenue) AS total_net_revenue, + SUM(oi.line_gross_margin) AS total_gross_margin, + SUM(oi.line_gross_margin) / NULLIF(SUM(oi.line_net_revenue), 0) AS avg_margin_pct, + AVG(oi.discount_pct) AS avg_discount_applied, + CAST(SUM(CASE WHEN oi.order_status = 'delivered' THEN 1 ELSE 0 END) AS DOUBLE) + / NULLIF(CAST(COUNT(*) AS DOUBLE), 0) AS reliability_score + FROM silver_confirmed_order_items oi + GROUP BY oi.supplier_id, oi.supplier_name, oi.supplier_country, oi.lead_time_days + ), + products_by_supplier AS ( + SELECT supplier_id, COUNT(*) AS products_sold + FROM (SELECT DISTINCT supplier_id, product_id FROM silver_confirmed_order_items) + GROUP BY supplier_id + ), + orders_count_by_supplier AS ( + SELECT supplier_id, COUNT(*) AS orders_fulfilled + FROM (SELECT DISTINCT supplier_id, order_id FROM silver_confirmed_order_items) + GROUP BY supplier_id + ) + SELECT + oi.supplier_id, oi.supplier_name, oi.supplier_country, oi.lead_time_days, + p.products_sold, + o.orders_fulfilled, + oi.total_units_sold, + oi.total_net_revenue, + oi.total_gross_margin, + oi.avg_margin_pct, + oi.avg_discount_applied, + oi.reliability_score, + inv.total_current_stock, + inv.total_sold AS inventory_units_sold, + inv.total_restocked AS inventory_units_restocked, + inv.total_returned AS inventory_units_returned + FROM orders_by_supplier oi + JOIN silver_inventory_by_supplier inv ON oi.supplier_id = inv.supplier_id + JOIN products_by_supplier p ON oi.supplier_id = p.supplier_id + JOIN orders_count_by_supplier o ON oi.supplier_id = o.supplier_id + """) + + +@dp.materialized_view( + name="gold_inventory_risk", + comment="Stock risk scoring per product.", + refresh_policy="incremental", +) +def gold_inventory_risk(): + return spark.sql(""" + SELECT + inv.product_id, inv.product_name, inv.category, inv.brand, + inv.supplier_name, inv.lead_time_days, + inv.total_stock_all_warehouses, + sales.units_sold, sales.avg_daily_units, + CASE WHEN sales.avg_daily_units > 0 + THEN inv.total_stock_all_warehouses / sales.avg_daily_units ELSE NULL END AS days_of_stock_remaining, + CASE + WHEN sales.avg_daily_units > 0 AND (inv.total_stock_all_warehouses / sales.avg_daily_units) < inv.lead_time_days * 1.5 THEN 'CRITICAL' + WHEN sales.avg_daily_units > 0 AND (inv.total_stock_all_warehouses / sales.avg_daily_units) < inv.lead_time_days * 3.0 THEN 'WARNING' + ELSE 'OK' + END AS stock_risk_level, + sales.net_revenue, sales.gross_margin + FROM ( + SELECT product_id, product_name, category, brand, supplier_name, lead_time_days, + SUM(current_stock) AS total_stock_all_warehouses + FROM silver_inventory_current + GROUP BY product_id, product_name, category, brand, supplier_name, lead_time_days + ) inv + JOIN ( + SELECT product_id, + SUM(quantity) AS units_sold, + SUM(quantity) / NULLIF(COUNT(DISTINCT date_trunc('day', order_created_at)), 0) AS avg_daily_units, + SUM(line_net_revenue) AS net_revenue, + SUM(line_gross_margin) AS gross_margin + FROM silver_confirmed_order_items + GROUP BY product_id + ) sales ON inv.product_id = sales.product_id + """) + + +@dp.materialized_view( + name="gold_realtime_inventory_alerts", + comment="CRITICAL stock filter.", + refresh_policy="incremental", +) +def gold_realtime_inventory_alerts(): + return spark.sql( + "SELECT * FROM gold_inventory_risk WHERE stock_risk_level = 'CRITICAL'" + ) + + +@dp.materialized_view( + name="gold_weekly_revenue_trend", + comment="Weekly revenue with WoW change, moving avg, YTD (window functions).", + refresh_policy="incremental", +) +def gold_weekly_revenue_trend(): + return spark.sql(""" + SELECT + week_start, category, + weekly_net_revenue, weekly_gross_margin, order_count, units_sold, + weekly_net_revenue - LAG(weekly_net_revenue, 1) OVER (PARTITION BY category ORDER BY week_start) + AS revenue_wow_change, + (weekly_net_revenue - LAG(weekly_net_revenue, 1) OVER (PARTITION BY category ORDER BY week_start)) + / NULLIF(LAG(weekly_net_revenue, 1) OVER (PARTITION BY category ORDER BY week_start), 0) + AS revenue_wow_pct_change, + AVG(weekly_net_revenue) OVER (PARTITION BY category ORDER BY week_start RANGE BETWEEN INTERVAL 3 WEEKS PRECEDING AND CURRENT ROW) + AS revenue_4wk_moving_avg, + AVG(weekly_gross_margin) OVER (PARTITION BY category ORDER BY week_start RANGE BETWEEN INTERVAL 3 WEEKS PRECEDING AND CURRENT ROW) + AS margin_4wk_moving_avg, + SUM(weekly_net_revenue) OVER (PARTITION BY category, EXTRACT(YEAR FROM week_start) ORDER BY week_start RANGE UNBOUNDED PRECEDING) + AS cumulative_ytd_revenue + FROM ( + SELECT + date_trunc('week', oi.order_created_at) AS week_start, + oi.category, + SUM(oi.line_net_revenue) AS weekly_net_revenue, + SUM(oi.line_gross_margin) AS weekly_gross_margin, + COUNT(DISTINCT oi.order_id) AS order_count, + SUM(oi.quantity) AS units_sold + FROM silver_confirmed_order_items oi + GROUP BY date_trunc('week', oi.order_created_at), oi.category + ) + """) + + +@dp.materialized_view( + name="gold_cancellation_impact", + comment="Cancellation rate windows (running totals).", + refresh_policy="incremental", +) +def gold_cancellation_impact(): + return spark.sql(""" + SELECT + category, week_start, + weekly_cancelled_orders, weekly_total_orders, + weekly_cancelled_revenue, weekly_total_revenue, + CAST(SUM(weekly_cancelled_orders) OVER (PARTITION BY category ORDER BY week_start RANGE UNBOUNDED PRECEDING) AS DOUBLE) + / NULLIF(CAST(SUM(weekly_total_orders) OVER (PARTITION BY category ORDER BY week_start RANGE UNBOUNDED PRECEDING) AS DOUBLE), 0) + AS cumulative_cancellation_rate, + SUM(weekly_cancelled_revenue) OVER (PARTITION BY category ORDER BY week_start RANGE UNBOUNDED PRECEDING) + AS cumulative_cancelled_revenue, + CAST(SUM(weekly_cancelled_orders) OVER (PARTITION BY category ORDER BY week_start RANGE BETWEEN INTERVAL 3 WEEKS PRECEDING AND CURRENT ROW) AS DOUBLE) + / NULLIF(CAST(SUM(weekly_total_orders) OVER (PARTITION BY category ORDER BY week_start RANGE BETWEEN INTERVAL 3 WEEKS PRECEDING AND CURRENT ROW) AS DOUBLE), 0) + AS cancellation_rate_4wk + FROM ( + SELECT + oi.category, + date_trunc('week', oi.order_created_at) AS week_start, + COUNT(DISTINCT CASE WHEN oi.order_status = 'cancelled' THEN oi.order_id END) AS weekly_cancelled_orders, + COUNT(DISTINCT oi.order_id) AS weekly_total_orders, + SUM(CASE WHEN oi.order_status = 'cancelled' THEN oi.line_net_revenue ELSE 0 END) AS weekly_cancelled_revenue, + SUM(oi.line_net_revenue) AS weekly_total_revenue + FROM silver_order_items_enriched oi + GROUP BY oi.category, date_trunc('week', oi.order_created_at) + ) + """) + + +@dp.materialized_view( + name="gold_product_demand_surge", + comment="Trailing-24h cart velocity vs stock & lead time (sliding window).", + refresh_policy="auto", +) +def gold_product_demand_surge(): + return spark.sql(""" + WITH recent_demand AS ( + SELECT + ce.product_id, + SUM(CASE WHEN ce.event_type = 'product_view' THEN 1 ELSE 0 END) AS recent_views, + SUM(CASE WHEN ce.event_type = 'add_to_cart' THEN 1 ELSE 0 END) AS recent_add_to_carts, + SUM(CASE WHEN ce.event_type = 'begin_checkout' THEN 1 ELSE 0 END) AS recent_checkouts + FROM silver_enriched_clickstream ce + WHERE ce.product_id IS NOT NULL + AND ce.event_timestamp > + (SELECT MAX(event_timestamp) FROM silver_enriched_clickstream) - INTERVAL 1 DAY + GROUP BY ce.product_id + ), + stock_by_product AS ( + SELECT product_id, + SUM(current_stock) AS total_stock_all_warehouses, + MAX(lead_time_days) AS lead_time_days + FROM silver_inventory_current + GROUP BY product_id + ) + SELECT + d.product_id, p.product_name, p.category AS product_category, p.brand AS product_brand, + d.recent_views, d.recent_add_to_carts, d.recent_checkouts, + COALESCE(s.total_stock_all_warehouses, 0) AS total_stock_all_warehouses, + s.lead_time_days, + CAST(GREATEST(COALESCE(s.total_stock_all_warehouses, 0), 0) AS DOUBLE) + / NULLIF(CAST(d.recent_add_to_carts AS DOUBLE), 0) AS days_of_stock_cover, + CAST(d.recent_checkouts AS DOUBLE) + / NULLIF(CAST(d.recent_views AS DOUBLE), 0) AS view_to_checkout_rate, + CASE + WHEN d.recent_add_to_carts = 0 THEN 'OK' + WHEN COALESCE(s.total_stock_all_warehouses, 0) <= 0 THEN 'SURGE_STOCKOUT_RISK' + WHEN CAST(GREATEST(s.total_stock_all_warehouses, 0) AS DOUBLE) / CAST(d.recent_add_to_carts AS DOUBLE) + < COALESCE(s.lead_time_days, 7) THEN 'SURGE_STOCKOUT_RISK' + WHEN CAST(GREATEST(s.total_stock_all_warehouses, 0) AS DOUBLE) / CAST(d.recent_add_to_carts AS DOUBLE) + < COALESCE(s.lead_time_days, 7) * 2 THEN 'WARNING' + ELSE 'OK' + END AS demand_alert + FROM recent_demand d + LEFT JOIN silver_products p ON d.product_id = p.product_id + LEFT JOIN stock_by_product s ON d.product_id = s.product_id + """) diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/part3.md b/docs.feldera.com/docs/use_cases/medallion_architecture/part3.md index 030ace38d67..e304f3f6e27 100644 --- a/docs.feldera.com/docs/use_cases/medallion_architecture/part3.md +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/part3.md @@ -51,3 +51,5 @@ Some of the silver views could be made incremental by hand by a data engineer More complex views, such as `gold_weekly_revenue_trend`, would require merge logic that is difficult to validate and entails significant engineering effort. With Feldera, you use the same SQL your batch engine runs now, get the same results, and pay a cost that tracks change size instead of data size. In addition, you get millisecond update latency across your full medallion architecture. + +Continue on to [Part 4](./part4.md) for a comparison between Databricks' IVM offering (Lakeflow Declarative Pipelines) and Feldera. \ No newline at end of file diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/part4.md b/docs.feldera.com/docs/use_cases/medallion_architecture/part4.md new file mode 100644 index 00000000000..f34a3266b5d --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/part4.md @@ -0,0 +1,305 @@ +# Part 4: Compare Feldera and Databricks Lakeflow Declarative Pipelines (Enzyme) + +Part [1](./part1.md) deploys the Feldera pipeline, which reads from a public bucket. [Part 2](./part2.md) demonstrates change data processing by pushing data to the Feldera pipeline via HTTP. [Part 3](./part3.md) compares Feldera and Spark, showcasing the difference between incremental view maintenance and batch refreshes. Part 4 compares Feldera with Databricks Enzyme engine. In this section, you will deploy the Databricks assets into **your own** workspace +and data in **your own** S3 bucket, then repoint the Feldera pipeline at your bucket bucket. Two things +run against the same Bronze Delta tables: + +- A **Lakeflow Declarative Pipeline** powered by Databricks Enzyme engine, maintains the Silver and Gold layers as Databricks + materialized views. Lakeflow Declarative Pipelines are the Databricks implementation of Incremental View Maintenance. The Declarative Pipeline uses the same SQL as the Feldera Pipeline. To fairly compare what Databricks **can** incrementalize, the [`refresh_policy`](https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-create-materialized-view-refresh-policy) for views in the pipeline is set to `incremental` where possible. Databricks currently cannot incrementalize the `gold_product_demand_surge` view, so the policy for this view is `auto`. +- An **`ingest_bronze` job** acts as a **data-generating process**: each run writes one + hour of CDC into the Bronze Delta tables. Point Feldera at those same tables and it + follows the Delta log, updating Gold sub-second after each write. The + Databricks pipeline runs immediately after ingestion. + +`ingest_bronze` stands in for a real operational source continually landing data. Instead +of `push_changes.py` pushing JSON to Feldera's HTTP ingress (Part 2), the writer here is a +Databricks job writing to Delta on S3, and Feldera consumes those Delta commits directly. + +## Files + +All four assets live in +[`medallion_architecture/databricks/`](https://github.com/feldera/feldera/tree/main/docs.feldera.com/docs/use_cases/medallion_architecture/databricks). + +| File | Role | +|---|---| +| `silver_gold_pipeline.py` | Pipeline source notebook — 8 Silver + 7 Gold materialized views (Bronze exposed as views). | +| `pipeline.yaml` | Settings for the Lakeflow pipeline that runs `silver_gold_pipeline.py` (Step 1). | +| `ingest_bronze.py` | Job notebook — optional `clean_start` re-seed, then apply one hour of CDC to the Bronze Delta tables. | +| `job.yaml` | Settings for the `compute_silver_gold` job (Step 2). | +| `measure_latency.py` | Reports Feldera's per-connector processing and end-to-end latency for changes (Step 4). | + +## Prerequisites +To run this demo yourself, you will need: + +- A Databricks workspace with **Unity Catalog** and **serverless** enabled. +- Read access to the read-only source bucket `s3://feldera-demos/ecommerce-cdc-{scale}` + (the snapshot and CDC files, anonymous read in `us-west-1`). +- **A writable S3 bucket you provision** for the working Bronze tables. The Databricks + job writes here and the Feldera pipeline reads from here, so both Databricks and Feldera + need access to it. This bucket is yours — you cannot write to `s3://feldera-demos`. + +## Storage layout + +| Layer | Location | Written by | +|---|---|---| +| Source snapshot + CDC (read-only) | `s3://feldera-demos/ecommerce-cdc-{scale}/{snapshot,cdc}` || +| Working Bronze (read/write) | `s3:///ecommerce-pipeline-{scale}/bronze` | `ingest_bronze.py` | +| Silver / Gold materialized views | Unity Catalog `{catalog}.{schema}` | the Lakeflow pipeline, `silver_gold_pipeline.py`| + +`{scale}` is the `scale_factor` with the dot replaced by a dash, e.g. `0.01` → `0-01`. The +working Bronze is kept separate from the read-only source so a `clean_start` can rebuild it +from scratch without touching the source snapshot. + +## Step 1 — Create the Lakeflow pipeline + +Create the pipeline first: the job's second task references it by ID. + +1. **Import the pipeline source.** Workspace → your folder → **Import** → add + `silver_gold_pipeline.py`. It imports as a notebook (the `# Databricks notebook source` + header makes the `# COMMAND ----------` cells render). +2. **Create the pipeline.** Sidebar → **Jobs & Pipelines** → **Create → ETL pipeline** + (Lakeflow Declarative Pipeline): + - **Source code**: the imported `silver_gold_pipeline.py`. + - **Serverless**: on. + - **Destination**: Unity Catalog → set **catalog** and **schema** where the Silver/Gold + materialized views publish (e.g. `main` / `ecommerce_demo`). + - **Advanced → Configuration**: add the keys the notebook reads via `spark.conf`: + - `scale_factor` = `0.01` + - `warehouse_bucket` = `s3://` + - The full settings are in [**`pipeline.yaml`**](https://github.com/feldera/feldera/tree/main/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml) (paste into *Settings → YAML*, or + `POST /api/2.0/pipelines`, instead of filling the form by hand). +3. **Save and copy the pipeline ID** (Pipeline details, or the URL) — Step 2 needs it. You + don't have to run it standalone; the job triggers it. + +## Step 2 — Upload `ingest_bronze.py` and create the job + +1. **Import the ingest notebook.** Workspace → **Import** → add `ingest_bronze.py`. +2. **Create the job.** Sidebar → **Jobs & Pipelines** → **Create job**, name it + `compute_silver_gold`. Full settings are in [**`job.yaml`**](https://github.com/feldera/feldera/tree/main/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml) (or use *Edit as YAML* / the + jobs API). Configure: + - **Job parameters**: `hour` = `""`, `clean_start` = `false`, `scale_factor` = `0.01`, + `source_bucket` = `s3://feldera-demos`, `warehouse_bucket` = `s3://`. + - **Task 1 — `ingest_bronze`** (Notebook): select `ingest_bronze.py`, compute + **Serverless**. Under the task **Parameters**, map each to the job parameter: + `hour` = `{{job.parameters.hour}}`, `clean_start` = `{{job.parameters.clean_start}}`, + `scale_factor` = `{{job.parameters.scale_factor}}`, + `source_bucket` = `{{job.parameters.source_bucket}}`, + `warehouse_bucket` = `{{job.parameters.warehouse_bucket}}`. + - **Task 2 — `refresh_silver_gold`** (Pipeline): **Depends on** `ingest_bronze`; select + the pipeline from Step 1 (paste its **pipeline ID** into `job.yaml`'s + `pipeline_task.pipeline_id`). Full refresh off. + +`scale_factor` and `warehouse_bucket` appear in both the pipeline configuration (Step 1) +and the job parameters (Step 2). The job passes them to the ingest notebook and, because +the keys match the pipeline's configuration, overrides the pipeline's values at run time — +so keep the two in sync when you change them. + +## Step 3 — `ingest_bronze` as a data-generating process + +Each run of the `ingest_bronze` task writes one hour of change into the Bronze Delta tables +under `s3:///ecommerce-pipeline-{scale}/bronze`: + +- `bronze_orders` carries status updates (delete-of-old + insert-of-new), so the hour's + rows are deduped to the latest `updated_at` per `order_id` and **merged**. +- `bronze_order_items`, `bronze_clickstream_events`, and `bronze_inventory_events` are + append-only event streams, so the hour's rows are **appended**. + +Every run commits new versions to the Delta log. That log is exactly what Feldera follows. + +**Run now → Run with parameters:** + +- **First run** — seed Bronze from the source snapshot into your bucket, then the + materialized views compute on the refresh step: + `clean_start = true`. +- **Each later run** — write ONE hour of CDC, in order: + `hour = 2025-11-30T00` (`clean_start = false`), then `2025-11-30T01`, and so on. + +The hours available are 7 days × 24, starting with `2025-11-30T00`. Run hours in order, once each — see [Notes](#notes). + +## Step 4 — Point the Feldera pipeline at your bucket + +The Feldera pipeline from Part 1 reads the Bronze snapshot from the public demo bucket in +`snapshot` mode. Repoint it at the Bronze tables that `ingest_bronze` writes and switch to +`snapshot_and_follow`, so Feldera loads the seeded snapshot and then follows the Delta log +as each `ingest_bronze` run commits a new hour. See the +[Delta input connector](/connectors/sources/delta) reference for the mode options. + +Edit each Bronze table's connector in the pipeline SQL. For the four fact tables that +receive CDC, change the `uri` to your bucket and set `mode` to `snapshot_and_follow`: + +```json +{ + "transport": { + "name": "delta_table_input", + "config": { + "uri": "s3:///ecommerce-pipeline-0-01/bronze/bronze_orders", + "mode": "snapshot_and_follow", + "aws_region": "", + "transaction_mode": "catchup" + } + } +} +``` + +| Bronze table | Mode | Why | +|---|---|---| +| `bronze_orders`, `bronze_order_items`, `bronze_clickstream_events`, `bronze_inventory_events` | `snapshot_and_follow` | Receive CDC every `ingest_bronze` run — follow the Delta log. | +| `bronze_suppliers`, `bronze_products`, `bronze_customers` | `snapshot` | Dimensions, seeded once and not updated by CDC. | + +Point every table's `uri` at `s3:///ecommerce-pipeline-0-01/bronze/` +(the `0-01` segment is `scale_factor` `0.01` with the dot replaced by a dash). + +### Configure AWS authentication for a private bucket + +The demo bucket is public, so Parts 1–3 use `"aws_skip_signature": "true"`. If your +bucket is private, **drop `aws_skip_signature`** and supply credentials instead. The +simplest option is an access key on the connector config (`aws_region` is required — the +Delta library does not auto-detect it): + +```json +"aws_access_key_id": "", +"aws_secret_access_key": "", +"aws_region": "" +``` + +For all supported options — access keys, session tokens, instance/container roles, KMS +encryption, custom endpoints — see +[configuring AWS authentication for the Delta connector](/connectors/sources/delta#storage-parameters) +and the [Setting AWS credentials example](/connectors/sources/delta#example-setting-aws-credentials). + +### Run order + +1. Run `ingest_bronze` once with `clean_start = true` (Step 3, first run) to seed Bronze in + your bucket. Feldera's `snapshot_and_follow` needs the snapshot to exist before it + starts. +2. Start (or restart) the Feldera pipeline. It backfills the seeded snapshot, then begins + following the Delta log. +3. Run `ingest_bronze` for each later hour. As each run commits, Feldera ingests the change + and updates Gold sub-second. Watch a Gold view from the **Ad-hoc query** or + **Change stream** tab, exactly as in [Part 2](./part2.md#watch-a-gold-view-update-in-real-time). +4. Measure the latency. Run `measure_latency.py` against the pipeline to see, per Bronze + Delta connector, how long Feldera took to ingest and fully process the latest Delta + commit (see [Result: Feldera's latency](#result-feldera) below): + + ```bash + uv pip install feldera python-dotenv + # or + pip install feldera python-dotenv + ``` + + Then run: + + ```bash + uv run measure_latency.py --pipeline ecommerce-medallion-architecture + # or + python measure_latency.py --pipeline ecommerce-medallion-architecture + ``` + +## Notes {#notes} + +- **Run hours in order, once each.** The append-only tables use plain appends (the cheapest + Delta write, and what keeps them insert-only), so re-running an already-applied hour + double-inserts into them. `bronze_orders` is idempotent (MERGE). For replayable hours, + switch the append tables to an idempotent MERGE on their primary key or use a + `txnAppId`/`txnVersion` marker on the append. +- **`scale_factor` and `warehouse_bucket` are set in two places** — the job parameters + (drive the Bronze S3 path in `ingest_bronze.py`) and the pipeline `configuration` (drives + the path the materialized views read from). The job values override the pipeline's at run + time via matching keys; keep the defaults in sync. +- **Databricks incremental refresh is best-effort.** Databricks (Enzyme) incrementalizes a + materialized view when the query shape allows and otherwise falls back to a full + recompute — unlike Feldera, which maintains the entire DAG incrementally on every commit. + That difference is the point of the comparison. + +## Result: the Databricks refresh for the first CDC hour + +The numbers below are the `refresh_silver_gold` pipeline task rebuilding the Silver and Gold +materialized views defined in `silver_gold_pipeline.py` after the first CDC hour +(`2025-11-30T00`) lands in Bronze. Databricks reports each update as a sequence of phases; +these are the phases for that single refresh: + +| Phase | Time | What it is | +|---|---|---| +| Created | 1s | Update queued. | +| Waiting for resources | 8s | Serverless compute provisioning. | +| Initializing | 18s | Pipeline graph + environment setup. | +| Setting up tables | 1s | Reconciling materialized-view metadata. | +| **Running** | **58s** | **Refreshing the 15 Silver/Gold views.** | +| **Total** | **86s** | End-to-end, data-landed to Gold-current. | + +Read the phases as two distinct costs: + +- **Compute — 58s.** The actual work of refreshing the views for *one hour* of change. + Work is not proportional to the size of the change. The clean start with full recomputation took about as long as the incremental run. In addition, `gold_product_demand_surge` full + recomputes every run (see below), and every Lakeflow flow carries fixed per-update + overhead. This is the same SQL Feldera runs. +- **Freshness — 86s.** What a consumer waits from the moment Bronze receives the hour to + the moment Gold reflects it. Serverless re-provisions and re-initializes on every + triggered run, so the 28s of `Waiting for resources` + `Initializing` is paid again each + time. + +Feldera pays neither cost the same way. It is always on, so there is no per-run +provisioning, and it maintains every view incrementally, so the update is proportional to +the size of the change. Updates are reflected in less than a second for one CDC hour, not 58s. + +### The view Databricks cannot incrementalize + +`gold_product_demand_surge` compares each clickstream event against a trailing 24-hour +window measured from the latest event: + +```sql +WHERE ce.event_timestamp > + (SELECT MAX(event_timestamp) FROM silver_enriched_clickstream) - INTERVAL 1 DAY +``` + +Databricks' Enzyme incremental planner cannot handle a `SUBQUERY_EXPRESSION` inside a Filter/And operator. +Feldera maintains identical SQL, and any other SQL you might write incrementally. + +You can confirm which views incrementalized and which full recomputed in the pipeline's +**event log** (each flow reports its refresh type per update). + +## Result: Feldera's latency for the first CDC hour {#result-feldera} + +The same first CDC hour (`2025-11-30T00`), measured on the Feldera side. When +`ingest_bronze` commits the hour, Feldera follows the Delta log and propagates the change +through every Silver and Gold view. `measure_latency.py` reads the `completed_frontier` +timestamps each Delta connector exposes on `/stats` and reports three numbers per Bronze +table (see [measuring end-to-end latency with input frontiers](/pipelines/latency/#measuring-end-to-end-latency-with-input-frontiers)): + +- **ingest→proc** — from the change arriving at the connector to the IVM engine finishing it. +- **proc→done** — from processing to the outputs reaching all sinks. +- **e2e** — the full ingest-to-sink latency (their sum). + +`version` is the Bronze table's Delta version; version `1` is the first CDC commit on top +of the seeded snapshot: + +``` +table version ingest→proc proc→done e2e +---------------------------------------------------------------------- +bronze_clickstream_events 1 0.08s 0.05s 0.13s +bronze_inventory_events 1 0.08s 0.05s 0.13s +bronze_orders 1 0.29s 0.08s 0.37s +bronze_order_items 1 0.08s 0.07s 0.15s +``` + +Feldera reflected the first CDC hour end-to-end in **0.37s at most** (`bronze_orders`, whose +delete-and-insert updates cost more than the append-only streams) against Databricks' 86s +for the identical SQL. There is no per-run startup: Feldera is already running, so the only +cost is processing the change itself. + +## Feldera vs. the Databricks Lakeflow Declarative Pipelines + +| | Databricks workflow | Feldera | +|---|---|---| +| Silver / Gold | Materialized views, incremental *where possible* (14/15) | All 15 views, always incremental | +| `gold_product_demand_surge` | Full recompute every refresh | Incremental | +| Orchestration | Job schedule + pipeline refresh | None — change-driven | +| Per-run startup | ~28s serverless provisioning | None — always on | +| Freshness for the first CDC hour | ~86s end-to-end | 0.37s at most | + +Both consume the identical Bronze Delta tables in your bucket and run the identical Silver +and Gold SQL. Databricks maintains 14 of the 15 views incrementally and full recomputes +`gold_product_demand_surge` on every refresh, taking 86s end-to-end (58s of it compute) to +reflect the first CDC hour. Feldera follows the Delta log and maintains every view +incrementally, reflecting the same hour end-to-end in 0.37s at most, without any per-run +startup. From 0308242a29c2828b13a9e04d5e16358ea0d65c9e Mon Sep 17 00:00:00 2001 From: "release-feldera-feldera[bot]" Date: Fri, 10 Jul 2026 22:22:31 +0000 Subject: [PATCH 3/7] ci: Prepare for v0.320.0 --- Cargo.lock | 42 +++++++++++------------ Cargo.toml | 26 +++++++------- openapi.json | 2 +- python/dbt-feldera/pyproject.toml | 2 +- python/felderize/pyproject.toml | 2 +- python/pyproject.toml | 2 +- python/uv.lock | 4 +-- sql-to-dbsp-compiler/SQL-compiler/pom.xml | 2 +- 8 files changed, 41 insertions(+), 41 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d172886ccc2..8f8059a5e0e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3976,7 +3976,7 @@ dependencies = [ [[package]] name = "dbsp" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "arc-swap", @@ -4066,7 +4066,7 @@ dependencies = [ [[package]] name = "dbsp_adapters" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix", "actix-codec", @@ -4210,7 +4210,7 @@ dependencies = [ [[package]] name = "dbsp_nexmark" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "ascii_table", @@ -5194,7 +5194,7 @@ dependencies = [ [[package]] name = "fda" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "arrow", @@ -5246,7 +5246,7 @@ dependencies = [ [[package]] name = "feldera-adapterlib" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix-web", "anyhow", @@ -5280,7 +5280,7 @@ dependencies = [ [[package]] name = "feldera-buffer-cache" -version = "0.319.0" +version = "0.320.0" dependencies = [ "crossbeam-utils", "enum-map", @@ -5308,7 +5308,7 @@ dependencies = [ [[package]] name = "feldera-datagen" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "async-channel 2.5.0", @@ -5334,7 +5334,7 @@ dependencies = [ [[package]] name = "feldera-fxp" -version = "0.319.0" +version = "0.320.0" dependencies = [ "bytecheck", "dbsp", @@ -5354,7 +5354,7 @@ dependencies = [ [[package]] name = "feldera-iceberg" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "chrono", @@ -5377,7 +5377,7 @@ dependencies = [ [[package]] name = "feldera-ir" -version = "0.319.0" +version = "0.320.0" dependencies = [ "proptest", "proptest-derive", @@ -5389,7 +5389,7 @@ dependencies = [ [[package]] name = "feldera-macros" -version = "0.319.0" +version = "0.320.0" dependencies = [ "prettyplease", "proc-macro2", @@ -5399,7 +5399,7 @@ dependencies = [ [[package]] name = "feldera-observability" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix-http", "awc", @@ -5414,7 +5414,7 @@ dependencies = [ [[package]] name = "feldera-rest-api" -version = "0.319.0" +version = "0.320.0" dependencies = [ "chrono", "feldera-observability", @@ -5433,7 +5433,7 @@ dependencies = [ [[package]] name = "feldera-samply" -version = "0.319.0" +version = "0.320.0" dependencies = [ "crossbeam", "feldera-size-of", @@ -5469,7 +5469,7 @@ dependencies = [ [[package]] name = "feldera-sqllib" -version = "0.319.0" +version = "0.320.0" dependencies = [ "arcstr", "base58", @@ -5513,7 +5513,7 @@ dependencies = [ [[package]] name = "feldera-storage" -version = "0.319.0" +version = "0.320.0" dependencies = [ "anyhow", "crossbeam", @@ -5536,7 +5536,7 @@ dependencies = [ [[package]] name = "feldera-types" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix-web", "anyhow", @@ -8813,7 +8813,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" [[package]] name = "pipeline-manager" -version = "0.319.0" +version = "0.320.0" dependencies = [ "actix-cors", "actix-files", @@ -10023,7 +10023,7 @@ dependencies = [ [[package]] name = "readers" -version = "0.319.0" +version = "0.320.0" dependencies = [ "async-std", "csv", @@ -11722,7 +11722,7 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "sltsqlvalue" -version = "0.319.0" +version = "0.320.0" dependencies = [ "dbsp", "feldera-sqllib", @@ -12193,7 +12193,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "storage-test-compat" -version = "0.319.0" +version = "0.320.0" dependencies = [ "dbsp", "derive_more 1.0.0", diff --git a/Cargo.toml b/Cargo.toml index 3b433eb54a7..2985273bea9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [workspace.package] authors = ["Feldera Team "] -version = "0.319.0" +version = "0.320.0" license = "MIT OR Apache-2.0" homepage = "https://github.com/feldera/feldera" repository = "https://github.com/feldera/feldera" @@ -108,7 +108,7 @@ csv = "1.2.2" csv-core = "0.1.10" dashmap = "6.1.0" datafusion = "53.1" -dbsp = { path = "crates/dbsp", version = "0.319.0" } +dbsp = { path = "crates/dbsp", version = "0.320.0" } dbsp_nexmark = { path = "crates/nexmark" } deadpool-postgres = "0.14.1" # Feldera fork of delta-rs: upstream tag `rust-v0.32.3` + 2 patches, on branch @@ -133,20 +133,20 @@ erased-serde = "0.3.31" fake = "2.10" fastbloom = "0.14.0" fdlimit = "0.3.0" -feldera-buffer-cache = { version = "0.319.0", path = "crates/buffer-cache" } +feldera-buffer-cache = { version = "0.320.0", path = "crates/buffer-cache" } feldera-cloud1-client = "0.1.2" feldera-datagen = { path = "crates/datagen" } -feldera-fxp = { version = "0.319.0", path = "crates/fxp", features = ["dbsp"] } +feldera-fxp = { version = "0.320.0", path = "crates/fxp", features = ["dbsp"] } feldera-iceberg = { path = "crates/iceberg" } -feldera-observability = { version = "0.319.0", path = "crates/feldera-observability" } -feldera-macros = { version = "0.319.0", path = "crates/feldera-macros" } -feldera-sqllib = { version = "0.319.0", path = "crates/sqllib" } -feldera-storage = { version = "0.319.0", path = "crates/storage" } -feldera-types = { version = "0.319.0", path = "crates/feldera-types" } -feldera-rest-api = { version = "0.319.0", path = "crates/rest-api" } -feldera-ir = { version = "0.319.0", path = "crates/ir" } -feldera-adapterlib = { version = "0.319.0", path = "crates/adapterlib" } -feldera-samply = { version = "0.319.0", path = "crates/samply" } +feldera-observability = { version = "0.320.0", path = "crates/feldera-observability" } +feldera-macros = { version = "0.320.0", path = "crates/feldera-macros" } +feldera-sqllib = { version = "0.320.0", path = "crates/sqllib" } +feldera-storage = { version = "0.320.0", path = "crates/storage" } +feldera-types = { version = "0.320.0", path = "crates/feldera-types" } +feldera-rest-api = { version = "0.320.0", path = "crates/rest-api" } +feldera-ir = { version = "0.320.0", path = "crates/ir" } +feldera-adapterlib = { version = "0.320.0", path = "crates/adapterlib" } +feldera-samply = { version = "0.320.0", path = "crates/samply" } flate2 = "1.1.0" form_urlencoded = "1.2.0" fs_extra = "1.3.0" diff --git a/openapi.json b/openapi.json index 7e13c03fa31..bee0dc1721c 100644 --- a/openapi.json +++ b/openapi.json @@ -10,7 +10,7 @@ "license": { "name": "MIT OR Apache-2.0" }, - "version": "0.319.0" + "version": "0.320.0" }, "paths": { "/config/authentication": { diff --git a/python/dbt-feldera/pyproject.toml b/python/dbt-feldera/pyproject.toml index 679a73eb616..5abf506ade6 100644 --- a/python/dbt-feldera/pyproject.toml +++ b/python/dbt-feldera/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta" name = "dbt-feldera" readme = "README.md" description = "The dbt adapter for Feldera — DBSP-native incremental view maintenance" -version = "0.319.0" +version = "0.320.0" license = "MIT" requires-python = ">=3.10" authors = [ diff --git a/python/felderize/pyproject.toml b/python/felderize/pyproject.toml index 499153b71fe..db8477dd1c1 100644 --- a/python/felderize/pyproject.toml +++ b/python/felderize/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta" name = "felderize" readme = "README.md" description = "SQL dialect to Feldera SQL translator agent" -version = "0.319.0" +version = "0.320.0" license = "MIT" requires-python = ">=3.10" authors = [ diff --git a/python/pyproject.toml b/python/pyproject.toml index b97cbfe0742..f0f9beb5c52 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta" name = "feldera" readme = "README.md" description = "The feldera python client" -version = "0.319.0" +version = "0.320.0" license = "MIT" requires-python = ">=3.10,<3.14" authors = [ diff --git a/python/uv.lock b/python/uv.lock index 202b9f34cc5..c3f873082a4 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -9,7 +9,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-07-01T03:34:58.843132494Z" +exclude-newer = "2026-07-03T22:21:56.598744158Z" exclude-newer-span = "P1W" [[package]] @@ -273,7 +273,7 @@ wheels = [ [[package]] name = "feldera" -version = "0.319.0" +version = "0.320.0" source = { editable = "." } dependencies = [ { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, diff --git a/sql-to-dbsp-compiler/SQL-compiler/pom.xml b/sql-to-dbsp-compiler/SQL-compiler/pom.xml index b2216965cf1..3925e922dc0 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/pom.xml +++ b/sql-to-dbsp-compiler/SQL-compiler/pom.xml @@ -19,7 +19,7 @@ 1.43.0 2.22.0 3.1.12 - 0.319.0 + 0.320.0 1.28.0 From a7e4fd860eb57307ebe587dfba1651a95db5e250 Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Fri, 10 Jul 2026 13:58:45 -0700 Subject: [PATCH 4/7] [DBSP] Add a 'positive' operator, a variant of 'distinct' Signed-off-by: Mihai Budiu --- crates/dbsp/src/algebra/zset.rs | 23 ++ crates/dbsp/src/circuit/circuit_builder.rs | 4 + crates/dbsp/src/mono.rs | 14 + crates/dbsp/src/operator/distinct.rs | 28 +- crates/dbsp/src/operator/dynamic/distinct.rs | 407 ++++++++++++++++--- 5 files changed, 418 insertions(+), 58 deletions(-) diff --git a/crates/dbsp/src/algebra/zset.rs b/crates/dbsp/src/algebra/zset.rs index b4c1b11db22..21a93fee1d4 100644 --- a/crates/dbsp/src/algebra/zset.rs +++ b/crates/dbsp/src/algebra/zset.rs @@ -250,6 +250,11 @@ pub trait ZSet: IndexedZSet { /// negative, so the result can be zero even if the Z-set contains nonzero /// weights. fn weighted_count(&self, sum: &mut Self::R); + + /// Returns a Z-set that contains all elements with positive weights from + /// `self` with their weights preserved. + #[cfg(test)] + fn positive(&self) -> Self; } impl ZSet for Z @@ -266,6 +271,24 @@ where cursor.step_key(); } } + + #[cfg(test)] + fn positive(&self) -> Self { + let factories = self.factories(); + let mut builder = Self::Builder::with_capacity(&factories, self.key_count(), self.len()); + let mut cursor = self.cursor(); + + while cursor.key_valid() { + let weight = **cursor.weight(); + if weight > 0 { + builder.push_val_diff(cursor.val(), weight.erase()); + builder.push_key(cursor.key()); + } + cursor.step_key(); + } + + builder.done() + } } #[cfg(test)] diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs index fdc1a7937a6..91f59bccf5a 100644 --- a/crates/dbsp/src/circuit/circuit_builder.rs +++ b/crates/dbsp/src/circuit/circuit_builder.rs @@ -544,6 +544,10 @@ dyn_clone::clone_trait_object!(StreamMetadata); /// data. It sets each record's weight to 1 if it is positive and drops the /// others. /// +/// The "positive" operator on a Z-set keeps positive weights unchanged and +/// maps all other weights to 0. It differs from "distinct" only in that it +/// does not clamp positive weights to 1. +/// /// ## Join on equal keys /// /// A DBSP equi-join takes batches `a` and `b` as input, finds all pairs of a diff --git a/crates/dbsp/src/mono.rs b/crates/dbsp/src/mono.rs index 28a7f416e7f..ae72ce47ef2 100644 --- a/crates/dbsp/src/mono.rs +++ b/crates/dbsp/src/mono.rs @@ -762,6 +762,13 @@ where self.inner().dyn_distinct_mono(&factories).typed() } + #[track_caller] + pub fn positive(&self) -> Self { + let factories = DistinctFactories::new::(); + + self.inner().dyn_positive_mono(&factories).typed() + } + #[track_caller] pub fn hash_distinct(&self) -> Self { let factories = HashDistinctFactories::new::(); @@ -1244,6 +1251,13 @@ where self.inner().dyn_distinct_mono(&factories).typed() } + #[track_caller] + pub fn positive(&self) -> Self { + let factories = DistinctFactories::new::(); + + self.inner().dyn_positive_mono(&factories).typed() + } + #[track_caller] pub fn hash_distinct(&self) -> Self { let factories = HashDistinctFactories::new::(); diff --git a/crates/dbsp/src/operator/distinct.rs b/crates/dbsp/src/operator/distinct.rs index 320429d9142..e5a7a1cab10 100644 --- a/crates/dbsp/src/operator/distinct.rs +++ b/crates/dbsp/src/operator/distinct.rs @@ -22,14 +22,7 @@ where self.inner().dyn_stream_distinct(&factories).typed() } -} -impl Stream -where - C: Circuit, - Z: IndexedZSet, - Z::InnerBatch: Send, -{ /// Incrementally deduplicate input stream. /// /// This is an incremental version of the @@ -62,4 +55,25 @@ where self.inner().dyn_has_distinct(&factories).typed() } + + /// Incrementally filter out tuples with non-positive weights. + /// + /// Given a stream of changes to relation `A`, computes a stream of + /// changes to relation `A'` that contains every `(key, weight)` tuple of + /// `A` with `weight > 0`, with the weight unchanged. + /// + /// This operator is like [`distinct`](`Self::distinct`), except that it + /// preserves the weights of the retained tuples instead of clamping them + /// to 1. + #[cfg(not(feature = "backend-mode"))] + #[track_caller] + pub fn positive(&self) -> Stream + where + Z: crate::typed_batch::ZSet, + { + let factories = + crate::operator::dynamic::distinct::DistinctFactories::new::(); + + self.inner().dyn_positive(&factories).typed() + } } diff --git a/crates/dbsp/src/operator/dynamic/distinct.rs b/crates/dbsp/src/operator/dynamic/distinct.rs index 3187b11ee89..0993728c83b 100644 --- a/crates/dbsp/src/operator/dynamic/distinct.rs +++ b/crates/dbsp/src/operator/dynamic/distinct.rs @@ -14,7 +14,7 @@ use crate::{ DBData, Runtime, Timestamp, ZWeight, algebra::{ AddByRef, HasOne, HasZero, IndexedZSet, Lattice, OrdIndexedZSet, OrdIndexedZSetFactories, - PartialOrder, ZRingValue, + PartialOrder, }, circuit::{ Circuit, Scope, Stream, WithClock, @@ -49,6 +49,45 @@ use super::{MonoIndexedZSet, MonoZSet}; circuit_cache_key!(DistinctId(StreamId => Stream)); circuit_cache_key!(DistinctIncrementalId(StreamId => Stream)); +circuit_cache_key!(PositiveIncrementalId(StreamId => Stream)); + +/// Abstract semantics of the generic `distinct` operator. +pub trait DistinctSemantics: 'static { + /// Name of the incremental operator specialized for the root scope. + const ROOT_SCOPE_NAME: &'static str; + /// Name of the incremental operator for nested scopes. + const NESTED_SCOPE_NAME: &'static str; + + /// Maps the weight `w` of a tuple in the input to the weight of + /// the same tuple in the output. + fn output_weight(w: ZWeight) -> ZWeight; +} + +/// [`DistinctSemantics`] of the `distinct` operator: tuples with positive +/// weight get weight 1. +pub struct ZeroOrOneWeight; + +impl DistinctSemantics for ZeroOrOneWeight { + const ROOT_SCOPE_NAME: &'static str = "DistinctIncrementalTotal"; + const NESTED_SCOPE_NAME: &'static str = "DistinctIncremental"; + + fn output_weight(w: ZWeight) -> ZWeight { + if w > 0 { 1 } else { 0 } + } +} + +/// [`DistinctSemantics`] of the `positive` operator: tuples with positive +/// weight keep their weight. +pub struct PositiveWeight; + +impl DistinctSemantics for PositiveWeight { + const ROOT_SCOPE_NAME: &'static str = "PositiveIncrementalTotal"; + const NESTED_SCOPE_NAME: &'static str = "PositiveIncremental"; + + fn output_weight(w: ZWeight) -> ZWeight { + w.max(0) + } +} pub struct DistinctFactories { pub input_factories: Z::Factories, @@ -189,6 +228,13 @@ impl Stream { self.dyn_distinct(factories) } + pub fn dyn_positive_mono( + &self, + factories: &DistinctFactories, + ) -> Stream { + self.dyn_positive(factories) + } + pub fn dyn_hash_distinct_mono( &self, factories: &HashDistinctFactories, @@ -221,6 +267,13 @@ impl Stream { self.dyn_distinct(factories) } + pub fn dyn_positive_mono( + &self, + factories: &DistinctFactories::Time>, + ) -> Stream { + self.dyn_positive(factories) + } + pub fn dyn_hash_distinct_mono( &self, factories: &HashDistinctFactories::Time>, @@ -307,9 +360,42 @@ where }) } + /// See [`Stream::positive`]. + pub fn dyn_positive(&self, factories: &DistinctFactories) -> Stream + where + Z: IndexedZSet + Send, + { + let circuit = self.circuit(); + circuit.region("positive", || { + let stream = self.try_sharded_version(); + + circuit + .cache_get_or_insert_with(PositiveIncrementalId::new(stream.stream_id()), || { + stream.dyn_positive_inner(factories) + }) + .clone() + }) + } + pub fn dyn_distinct_inner(&self, factories: &DistinctFactories) -> Stream where Z: IndexedZSet + Send, + { + self.distinct_inner_generic::(factories) + .mark_distinct() + } + + pub fn dyn_positive_inner(&self, factories: &DistinctFactories) -> Stream + where + Z: IndexedZSet + Send, + { + self.distinct_inner_generic::(factories) + } + + fn distinct_inner_generic(&self, factories: &DistinctFactories) -> Stream + where + Z: IndexedZSet + Send, + S: DistinctSemantics, { let circuit = self.circuit(); @@ -318,7 +404,7 @@ where if circuit.root_scope() == 0 { // Use an implementation optimized to work in the root scope. circuit.add_binary_operator( - StreamingBinaryWrapper::new(DistinctIncrementalTotal::new( + StreamingBinaryWrapper::new(DistinctIncrementalTotal::<_, _, S>::new( Location::caller(), &factories.input_factories, )), @@ -339,7 +425,7 @@ where // └───────────┘ └─────┘ └─────┘ └───────────────────┘ // ``` circuit.add_binary_operator( - StreamingBinaryWrapper::new(DistinctIncremental::new( + StreamingBinaryWrapper::new(DistinctIncremental::<_, _, _, S>::new( Location::caller(), &factories.input_factories, &factories.aux_factories, @@ -356,7 +442,6 @@ where ) } .mark_sharded() - .mark_distinct() } } @@ -403,14 +488,15 @@ where } } -/// Incremental version of the distinct operator that only works in the -/// top-level scope (i.e., for totally ordered timestamps). +/// Incremental version of the distinct and positive operators that only works +/// in the top-level scope (i.e., for totally ordered timestamps). /// /// Takes a stream `a` of changes to relation `A` and a stream with delayed /// value of `A`: `z^-1(A) = a.integrate().delay()` and computes -/// `distinct(A) - distinct(z^-1(A))` incrementally, by only considering -/// values in the support of `a`. -struct DistinctIncrementalTotal { +/// `f(A) - f(z^-1(A))` incrementally, by only considering values in the +/// support of `a`, where `f` applies [`DistinctSemantics::output_weight`] to +/// the weight of every tuple. +struct DistinctIncrementalTotal { input_factories: Z::Factories, location: &'static Location<'static>, @@ -423,10 +509,10 @@ struct DistinctIncrementalTotal { chunk_size: usize, first_chunk_size: usize, - _type: PhantomData<(Z, I)>, + _type: PhantomData<(Z, I, S)>, } -impl DistinctIncrementalTotal { +impl DistinctIncrementalTotal { pub fn new(location: &'static Location<'static>, input_factories: &Z::Factories) -> Self { Self { input_factories: input_factories.clone(), @@ -470,13 +556,14 @@ impl DistinctIncrementalTotal { } } -impl Operator for DistinctIncrementalTotal +impl Operator for DistinctIncrementalTotal where Z: IndexedZSet, I: 'static, + S: DistinctSemantics, { fn name(&self) -> Cow<'static, str> { - Cow::from("DistinctIncrementalTotal") + Cow::from(S::ROOT_SCOPE_NAME) } fn location(&self) -> OperatorLocation { @@ -495,10 +582,11 @@ where } } -impl StreamingBinaryOperator>, I, Z> for DistinctIncrementalTotal +impl StreamingBinaryOperator>, I, Z> for DistinctIncrementalTotal where Z: IndexedZSet, I: WithSnapshot + 'static, + S: DistinctSemantics, { fn eval( self: Rc, @@ -555,15 +643,13 @@ where let new_weight = old_weight.add_by_ref(&w); - if old_weight.le0() { - // Weight changes from non-positive to positive. - if new_weight.ge0() && !new_weight.is_zero() { - builder.push_val_diff(v, ZWeight::one().erase()); - any_values = true; - } - } else if new_weight.le0() { - // Weight changes from positive to non-positive. - builder.push_val_diff(v, ZWeight::one().neg().erase()); + // Emit the change in the tuple's output weight. For + // `distinct` it is nonzero only when the integrated + // weight changes sign; for `positive` also when a + // positive weight changes value. + let diff = S::output_weight(new_weight) - S::output_weight(old_weight); + if !diff.is_zero() { + builder.push_val_diff(v, diff.erase()); any_values = true; } @@ -574,12 +660,15 @@ where delta_cursor.step_val(); } } else { + // The key is missing from the integral: the weight + // if copied from the delta while delta_cursor.val_valid() { let new_weight = **delta_cursor.weight(); debug_assert!(!new_weight.is_zero()); - if new_weight.ge0() { - builder.push_val_diff(delta_cursor.val(), ZWeight::one().erase()); + let output_weight = S::output_weight(new_weight); + if !output_weight.is_zero() { + builder.push_val_diff(delta_cursor.val(), output_weight.erase()); any_values = true; } @@ -612,7 +701,7 @@ where type KeysOfInterest = BTreeMap, R>>>; #[derive(SizeOf)] -struct DistinctIncremental +struct DistinctIncremental where Z: IndexedZSet, T: WithSnapshot, @@ -649,15 +738,16 @@ where chunk_size: usize, first_chunk_size: usize, - _type: PhantomData<(Z, T)>, + _type: PhantomData<(Z, T, S)>, } -impl DistinctIncremental +impl DistinctIncremental where Z: IndexedZSet, T: WithSnapshot, T::Batch: ZBatchReader, Clk: WithClock