## ----include = FALSE----------------------------------------------------------
# The raw-SQL examples need Python sqlglot, so only evaluate chunks when
# the full stack is available. has_sqlglot() initializes Python, which lets
# reticulate provision an environment over the network, so it must not be
# called during a CRAN check; there the chunks render as unevaluated code.
can_eval <- identical(Sys.getenv("NOT_CRAN"), "true") &&
  requireNamespace("dplyr", quietly = TRUE) &&
  requireNamespace("dbplyr", quietly = TRUE) &&
  requireNamespace("duckdb", quietly = TRUE) &&
  tryCatch(dplyneage::has_sqlglot(), error = function(e) FALSE)

knitr::opts_chunk$set(
  collapse = TRUE,
  comment = "#>",
  eval = can_eval
)

## ----eval = FALSE-------------------------------------------------------------
# pak::pak("tgerke/dplyneage")

## ----message=FALSE------------------------------------------------------------
library(dplyneage)
library(dplyr)
library(duckdb)

con <- dbConnect(duckdb(), ":memory:")

customers <- tibble(
  id = 1:5,
  name = c("Alice", "Bob", "Charlie", "Diana", "Eve"),
  email = paste0(tolower(name), "@example.com")
)

orders <- tibble(
  order_id = 1:10,
  customer_id = rep(1:5, each = 2),
  amount = c(100, 150, 200, 75, 300, 125, 180, 90, 250, 160)
)

copy_to(con, customers, "customers", overwrite = TRUE)
copy_to(con, orders, "orders", overwrite = TRUE)

## -----------------------------------------------------------------------------
tbl(con, "customers") |>
  select(id, name) |>
  extract_lineage() |>
  lineage_flow(height = "300px")

## -----------------------------------------------------------------------------
tbl(con, "customers") |>
  left_join(tbl(con, "orders"), by = c("id" = "customer_id")) |>
  group_by(id, name) |>
  summarise(total_spent = sum(amount, na.rm = TRUE), .groups = "drop") |>
  extract_lineage() |>
  lineage_flow(height = "400px")

## -----------------------------------------------------------------------------
extract_lineage("
  WITH recent AS (
    SELECT customer_id, amount FROM orders WHERE order_date > '2024-01-01'
  )
  SELECT customer_id, SUM(amount) AS total FROM recent GROUP BY customer_id
") |>
  lineage_flow(height = "300px")

## -----------------------------------------------------------------------------
extract_lineage("
  SELECT COALESCE(u.email, a.email) AS email
  FROM users u FULL JOIN archive a ON u.id = a.id
") |>
  lineage_flow(height = "300px")

## -----------------------------------------------------------------------------
extract_lineage(
  "SELECT * FROM customers",
  schema = list(customers = c("id", "name", "email"))
) |>
  lineage_flow(height = "300px")

## -----------------------------------------------------------------------------
extract_lineage(
  "SELECT c.name, order_date
   FROM customers c JOIN orders o ON c.id = o.customer_id",
  schema = list(
    customers = c("id", "name", "email"),
    orders = c("order_id", "customer_id", "order_date", "amount")
  )
) |>
  lineage_flow(height = "300px")

## ----eval = FALSE-------------------------------------------------------------
# extract_lineage(query, dialect = "postgres")
# extract_lineage(query, dialect = "snowflake")

## ----eval = can_eval && requireNamespace("RSQLite", quietly = TRUE)-----------
sales <- dbplyr::memdb_frame(
  customer_id = c(1, 1, 2),
  amount = c(100, 250, 40),
  .name = "sales"
)

sales |>
  group_by(customer_id) |>
  summarise(total = sum(amount, na.rm = TRUE)) |>
  extract_lineage() |>
  lineage_flow(height = "300px")

## -----------------------------------------------------------------------------
nodes <- list(
  create_table_node(
    table_name = "customers",
    columns = c("id", "name", "email"),
    x = 0, y = 100,
    table_type = "source"
  ),
  create_table_node(
    table_name = "customer_summary",
    columns = c("customer_id", "full_name", "contact"),
    x = 500, y = 100,
    table_type = "target"
  )
)

edges <- list(
  create_column_edge("customers", "id", "customer_summary", "customer_id"),
  create_column_edge("customers", "name", "customer_summary", "full_name"),
  create_column_edge("customers", "email", "customer_summary", "contact")
)

lineage_flow(nodes, edges, height = "300px")

## -----------------------------------------------------------------------------
lineage <- tbl(con, "customers") |>
  left_join(tbl(con, "orders"), by = c("id" = "customer_id")) |>
  group_by(id, name) |>
  summarise(total_spent = sum(amount, na.rm = TRUE), .groups = "drop") |>
  extract_lineage()

lineage_json(lineage)

## ----eval = can_eval && requireNamespace("igraph", quietly = TRUE)------------
path <- tempfile(fileext = ".graphml")
lineage_graphml(lineage, path)

g <- igraph::read_graph(path, format = "graphml")

# Everything upstream of total_spent
igraph::subcomponent(g, "output.total_spent", mode = "in")

# Everything downstream of orders.amount
igraph::subcomponent(g, "orders.amount", mode = "out")

## ----include = FALSE----------------------------------------------------------
dbDisconnect(con)