---
title: "Working with Fabric Eventhouses (real-time data)"
output: rmarkdown::html_vignette
vignette: >
  %\VignetteIndexEntry{Working with Fabric Eventhouses (real-time data)}
  %\VignetteEngine{knitr::rmarkdown}
  %\VignetteEncoding{UTF-8}
---

```{r, include = FALSE}
knitr::opts_chunk$set(collapse = TRUE, comment = "#>", eval = FALSE)
```

An Eventhouse stores event, log, and time-series data in KQL databases. In this
guide, *ingestion* means adding rows to a KQL table. Start with the discovered
database's `$write_table()` method (`fabric_kql_write_table()`), which accepts
an R data frame and manages staging, status checks, and cleanup for you.

After the basic write, this guide shows how to monitor ingestion, use an
existing storage file, scale up with Arrow, and export a large query result.
The lower-level ingestion routes used by these workflows are currently in
preview.

## Find a KQL database

Start with a discovered KQL database. The result is a read-only
`FabricKqlDatabase` R6 object. Its fields contain both the query and ingestion
URIs, and its methods reuse that context and the discovery credential:

```{r, eval = FALSE}
library(fabricQueryR)

database <- fabric_kql_databases("Telemetry workspace")[[1]]
database$ingestion_service_uri
```

## Write a small data frame in one call

Supply a destination table and an ordinary data frame or tibble. Use a new test
table while learning:

```{r, eval = FALSE}
written <- database$write_table(
  table = "Events",
  data = data.frame(
    id = 1:3,
    category = c("A", "B", "A"),
    amount = c(10.5, 20, 30.5)
  ),
  create_if_missing = TRUE,
  ingest_if_not_exists = "r-events-2026-08-14"
)

written$status$state
written$rows
written$staging_retained
```

The function writes temporary Parquet data, uploads it to Fabric, queues the
ingestion, and waits for a final status. With the default `cleanup = TRUE`,
service-owned Storage sources may be deleted after download, before ingestion
succeeds. OneLake staging is removed only after confirmed success. Set
`cleanup = FALSE` to retain Storage sources for recovery; a failed Storage batch
otherwise reports `staging_retained = NA` because source retention is unknown.
`create_if_missing = TRUE` creates a basic table from the R object's
columns when needed.

For an existing table, the source names and types must match. Supply a
predefined Parquet `mapping` or explicit `column_types` when inference is not
appropriate. See `?fabric_kql_write_table` for supported type mappings and
staging recovery options.

## Queue a tracked batch

Use the lower-level `$ingest()` method (`fabric_kql_ingest()`) when the source
file already exists in OneLake or supported blob storage. It does not upload a
local file or serialize an R object, and the destination table must already
exist.

The `mapping` argument is optional. When it is omitted, Kusto derives an
identity mapping from the existing table schema: ordered text formats such as
CSV map source columns by position, while JSON, Parquet, Avro, ORC, and
W3CLOGFILE map fields to case-sensitive table-column names. Use a validated
named mapping when source order or names differ, or when ingestion-time
transformations are required. Supply the complete source path and its format:

```{r, eval = FALSE}
source <- paste0(
  "https://onelake.dfs.fabric.microsoft.com/",
  "<workspace-id>/<lakehouse-id>",
  "/Files/events/2026-08-14.csv;impersonate"
)
```

The signed-in identity must be able to read the file. Do not print or log a
source string that contains credentials. Source IDs are generated when omitted
and remain available on the returned handle. This example uses identity mapping
unless `FABRIC_KQL_INGESTION_MAPPING` names a predefined CSV mapping:

```{r, eval = FALSE}
mapping <- Sys.getenv("FABRIC_KQL_INGESTION_MAPPING", unset = "")

ingestion <- database$ingest(
  table = "Events",
  sources = source,
  format = "csv",
  mapping = if (nzchar(mapping)) mapping else NULL,
  ignore_first_record = TRUE,
  tags = "source:daily-export",
  ingest_if_not_exists = "events-2026-08-14"
)

ingestion$id
ingestion$sources$source_id
```

Use a stable `ingest_if_not_exists` key when the same source file may be
submitted again. Idempotency keys require one source per call; submit multiple
files separately with a distinct stable key for each file. Queued ingestion is
an advanced, at-least-once workflow: after an uncertain network result, inspect
the tracked operation and target table before submitting the source again. See
`?fabric_kql_ingest` for batching, source deletion, and storage-authentication
options.

## Wait and inspect the outcome

One status snapshot with `$ingestion_status()`
(`fabric_kql_ingestion_status()`) is useful for a scheduler that persists
operation IDs:

```{r, eval = FALSE}
snapshot <- database$ingestion_status(ingestion)
snapshot$state
snapshot$counts
```

For an interactive or single-process batch, `$ingestion_wait()` calls
`fabric_kql_ingestion_status(..., wait = TRUE)` with a client-side deadline:

```{r, eval = FALSE}
result <- database$ingestion_wait(
  ingestion,
  timeout = 900,
  poll_interval = 2
)

result$state
result$details
```

The deadline stops only the R waiter; it does not cancel ingestion in Fabric.
By default, a failed or partially successful batch raises an R condition. To
inspect every terminal state as data instead, call `$ingestion_wait()`
(`fabric_kql_ingestion_status(..., wait = TRUE)`) with
`error_on_failure = FALSE`:

```{r, eval = FALSE}
result <- database$ingestion_wait(
  ingestion,
  error_on_failure = FALSE
)

failed <- subset(
  result$details,
  status %in% c("Failed", "Canceled")
)
failed[c("source_id", "error_code", "failure_status", "message")]
```

Query the destination after success with `$query()` (`fabric_kql_query()`):

```{r, eval = FALSE}
loaded <- database$query(
  query = "Events | where ingestion_time() > ago(1h) | take 100"
)
```

The caller needs permission to ingest into the destination table and read the
source file. Keep a manually staged source until the tracked result is final and
verified. With `$write_table()` (`fabric_kql_write_table()`), select
`cleanup = FALSE` if service-owned Storage sources must remain available until
you verify the result; the default permits deletion after download.

## Scale up with Arrow

For data larger than memory, pass an Arrow Dataset, Scanner, 'dplyr' query,
RecordBatchReader, or compatible stream to the same high-level writer:

```{r, eval = FALSE}
dataset <- arrow::open_dataset("local-parquet-directory")

written <- database$write_table(
  table = "Events",
  data = dataset,
  mapping = "EventsParquet"
)
```

Here `$write_table()` calls `fabric_kql_write_table()`, as in the data-frame
example above.

The source is processed in batches instead of first being collected into an R
data frame. A supplied RecordBatchReader is single-use.

## Export a large KQL result to OneLake

`$query()` (`fabric_kql_query()`) is the right interface when the result belongs
in R. When the result is too large for the client-result channel or should
remain in Fabric, `$export()` (`fabric_kql_export()`) runs Kusto's service-side
export and writes the first result set directly to storage:

```{r, eval = FALSE}
lakehouse <- fabric_lakehouses("Telemetry workspace")[[1]]

exported <- database$export(
  query = "Events | where amount > 0",
  destination = lakehouse,
  path = "Files/exports/events-positive-amount",
  format = "parquet",
  name_prefix = "events",
  compression_type = "snappy"
)

exported$state
exported$records
exported$artifacts
```

The signed-in identity needs write access to the destination. If an export
fails, treat any files already written as incomplete and inspect the returned
operation before starting a replacement export.