### Quick Start: Download Additional JARs (First Time)

Source: https://github.com/delta-io/delta/blob/master/flink/README.md

Downloads the AWS SDK bundle and Guava JARs into the Docker usrlib directory; required for S3 access and necessary for the first-time setup.

```bash
cd docker/<flink_version>/usrlib
chmod +x init.sh
wget https://repo1.maven.org/maven2/software/amazon/awssdk/bundle/2.23.19/bundle-2.23.19.jar 
wget https://repo1.maven.org/maven2/com/google/guava/guava/33.5.0-jre/guava-33.5.0-jre.jar
```

--------------------------------

### List of example main classes

Source: https://github.com/delta-io/delta/blob/master/examples/scala/README.md

Output shown when running `./build/sbt run` in the examples directory, listing the available main classes to select.

```text
Multiple main classes detected. Select one to run:
 [1] example.Quickstart
 [2] example.QuickstartSQL
 [3] example.QuickstartSQLOnPaths
 [4] example.Streaming
 [5] example.Utilities
```

--------------------------------

### SQL AS OF example with TIMESTAMP and VERSION

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-batch.mdx

Example queries using TIMESTAMP AS OF with a literal timestamp and VERSION AS OF with version 123 on a Delta table path.

```sql
SELECT * FROM default.people10m TIMESTAMP AS OF '2018-10-18T22:15:12.013Z'
SELECT * FROM delta.`/tmp/delta/people10m` VERSION AS OF 123
```

--------------------------------

### Quick Start: Start Local Flink Cluster with Docker Compose

Source: https://github.com/delta-io/delta/blob/master/flink/README.md

Starts a local Flink cluster consisting of one JobManager and multiple TaskManagers for testing and development.

```bash
cd docker/<flink_version>
docker compose up -d
```

--------------------------------

### Install dependencies with pnpm

Source: https://github.com/delta-io/delta/blob/master/docs/README.md

Install project dependencies using pnpm.

```bash
pnpm i
```

--------------------------------

### Install pnpm via corepack

Source: https://github.com/delta-io/delta/blob/master/docs/README.md

Install pnpm globally using corepack and enable it.

```bash
npm install --global corepack@latest
corepack enable pnpm
```

--------------------------------

### Initialize Engine and Get Table Schema

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-kernel-java.mdx

Creates an Engine instance, loads a table by path, gets the latest snapshot, and retrieves the schema. This is the first step in reading a Delta table.

```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.engine.*;

Engine myEngine = new MyEngine();
Table myTable = Table.forPath(myTablePath);
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
StructType mySchema = mySnapshot.getSchema(myEngine);
```

--------------------------------

### Install Node version with nvm

Source: https://github.com/delta-io/delta/blob/master/docs/README.md

Install Node v22.14.0 using nvm. Run this in the delta repo root.

```bash
nvm install
```

--------------------------------

### Scala: Read change data in streaming queries with readChangeFeed

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-change-data-feed.mdx

Scala equivalent of the Python examples, showing how to start a streaming read of change data with a starting version, a starting timestamp, or without either. Note the Scala example uses startingVersion for the timestamp case, which may be a typo in the source.

```scala
// providing a starting version
spark.readStream.format("delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("myDeltaTable")

// providing a starting timestamp
spark.readStream.format("delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", "2021-04-21 05:35:43")
  .load("/pathToMyDeltaTable")

// not providing a starting version/timestamp will result in the latest snapshot being fetched first
spark.readStream.format("delta")
  .option("readChangeFeed", "true")
  .table("myDeltaTable")
```

--------------------------------

### Remove action example

Source: https://github.com/delta-io/delta/blob/master/PROTOCOL.md

Example of a remove action in the Delta Lake transaction log. The path is a URI that needs to be decoded to get the data file path. The deletionTimestamp is in milliseconds since the epoch.

```json
{
  "remove": {
    "path": "part-00001-9…..snappy.parquet",
    "deletionTimestamp": 1515488792485,
    "baseRowId": 4071,
    "defaultRowCommitVersion": 41,
    "dataChange": true
  }
}
```

--------------------------------

### Python Delta Connect Client Example

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-spark-connect.mdx

Python example using the Delta Connect client. Requires pip install pyspark==4.0.0 and delta-spark==4.0.0. Pass a remote SparkSession to the DeltaTable API. The example retrieves a DeltaTable by name, shows its DataFrame, and updates rows where id is even by adding 100 to id.

```python
from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

deltaTable = DeltaTable.forName(spark, "my_table")
deltaTable.toDF().show()

deltaTable.update(
  condition = "id % 2 == 0",
  set = {"id": "id + 100"}
)
```

--------------------------------

### SHOW PARTITIONS SQL examples with outputs

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-utility/index.mdx

SQL commands to list partitions of a Delta table, optionally filter by a partition value, and specify a table by path. Outputs show typed columns for each partition key.

```sql
SHOW PARTITIONS eventsTable;
```

```text
+----------+-------+
| date     | region|
+----------+-------+
|2024-01-01|us     |
|2024-01-02|eu     |
+----------+-------+
```

```sql
SHOW PARTITIONS eventsTable PARTITION (date = '2024-01-01');
```

```text
+----------+------+
| date     |region|
+----------+------+
|2024-01-01|us    |
+----------+------+
```

```sql
SHOW PARTITIONS delta.`/data/events/`;
```

```text
+----------+-------+
| date     | region|
+----------+-------+
|2024-01-01|us     |
|2024-01-02|eu     |
+----------+-------+
```

--------------------------------

### Read changes since timestamp with startingTimestamp option (Scala)

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-streaming/index.mdx

Use the startingTimestamp option to start reading from a specific timestamp. This example reads changes since 2018-10-18 from the user_events table. The option takes effect only when starting a new streaming query; if a checkpoint exists, it is ignored.

```scala
spark.readStream.format("delta")
  .option("startingTimestamp", "2018-10-18")
  .load("/tmp/delta/user_events")
```

--------------------------------

### Generate docs with live preview and API docs

Source: https://github.com/delta-io/delta/blob/master/docs/README.md

Run the docs generation script with live HTML preview and API docs. Prints a URL to preview the docs.

```bash
python3 docs/generate_docs.py --livehtml --api-docs
```

--------------------------------

### Read changes since version 5 with startingVersion option (Scala)

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-streaming/index.mdx

Use the startingVersion option to start reading from a specific Delta Lake version. This example reads changes since version 5 from the user_events table. The option takes effect only when starting a new streaming query; if a checkpoint exists, it is ignored.

```scala
spark.readStream.format("delta")
  .option("startingVersion", "5")
  .load("/tmp/delta/user_events")
```

--------------------------------

### Create Dataproc Metastore with gcloud

Source: https://github.com/delta-io/delta/blob/master/benchmarks/README.md

Sample command to create a Dataproc Metastore service named 'dataproc-metastore-for-benchmarks' with enterprise tier. Replace <region> with your region.

```bash
gcloud metastore services create dataproc-metastore-for-benchmarks \
    --location=<region> \
    --tier=enterprise
```

--------------------------------

### Declare identity columns with DeltaTable.create

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-batch.mdx

Declares identity columns on a Delta table using the DeltaTable builder API. The example shows both generatedAlwaysAs and generatedByDefaultAs variants, with optional start and step parameters. Identity columns are supported in Delta Lake 3.3 and above, and declaring one disables concurrent transactions.

```python
from delta.tables import DeltaTable, IdentityGenerator
from pyspark.sql.types import LongType

DeltaTable.create()
  .tableName("table_name")
  .addColumn("id_col1", dataType=LongType(), generatedAlwaysAs=IdentityGenerator())
  .addColumn("id_col2", dataType=LongType(), generatedAlwaysAs=IdentityGenerator(start=-1, step=1))
  .addColumn("id_col3", dataType=LongType(), generatedByDefaultAs=IdentityGenerator())
  .addColumn("id_col4", dataType=LongType(), generatedByDefaultAs=IdentityGenerator(start=-1, step=1))
  .execute()
```

```scala
import io.delta.tables.DeltaTable
import org.apache.spark.sql.types.LongType

DeltaTable.create(spark)
  .tableName("table_name")
  .addColumn(
    DeltaTable.columnBuilder(spark, "id_col1")
      .dataType(LongType)
      .generatedAlwaysAsIdentity().build())
  .addColumn(
    DeltaTable.columnBuilder(spark, "id_col2")
      .dataType(LongType)
      .generatedAlwaysAsIdentity(start = -1L, step = 1L).build())
  .addColumn(
    DeltaTable.columnBuilder(spark, "id_col3")
      .dataType(LongType)
      .generatedByDefaultAsIdentity().build())
  .addColumn(
    DeltaTable.columnBuilder(spark, "id_col4")
      .dataType(LongType)
      .generatedByDefaultAsIdentity(start = -1L, step = 1L).build())
  .execute()
```

--------------------------------

### Create Table, Snapshot, and Scan with Delta Kernel

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-kernel.mdx

Steps to create a Table, Snapshot, and Scan for a Delta table using the default Engine. The first block shows imports and table creation; the second gets the latest snapshot; the third retrieves version and schema; the fourth builds a Scan and obtains scan state and scan files.

```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.*;
import org.apache.hadoop.conf.Configuration;

String myTablePath = <my-table-path>; // fully qualified table path. Ex: file:/user/tables/myTable
Configuration hadoopConf = new Configuration();
Engine myEngine = DefaultEngine.create(hadoopConf);
Table myTable = Table.forPath(myEngine, myTablePath);
```

```java
Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
```

```java
long version = mySnapshot.getVersion(myEngine);
StructType tableSchema = mySnapshot.getSchema(myEngine);
```

```java
Scan myScan = mySnapshot.getScanBuilder(myEngine).build()

// Common information about scanning for all data files to read.
Row scanState = myScan.getScanState(myEngine)

// Information about the list of scan files to read
CloseableIterator<FilteredColumnarBatch> scanFiles = myScan.getScanFiles(myEngine)
```

--------------------------------

### Scala example: Enable withEventTimeOrder on Delta source

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-streaming/index.mdx

Set the withEventTimeOrder option to 'true' on a Delta table stream source to process the initial snapshot in event time order, preventing data drop in stateful queries with a watermark. Must be set before starting the query; cannot be changed while initial snapshot is processing.

```scala
spark.readStream.format("delta")
  .option("withEventTimeOrder", "true")
  .load("/tmp/delta/user_events")
  .withWatermark("event_time", "10 seconds")
```

--------------------------------

### Prepare data and partition values for writing

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-kernel.mdx

Shows how to prepare the connector data as an iterator of FilteredColumnarBatch and create a partition value map. For un-partitioned tables, the map should be empty.

```java
// The data generated by the connector to write into a table
CloseableIterator<FilteredColumnarBatch> data = ...

// Create partition value map
Map<String, Literal> partitionValues =
  Collections.singletonMap(
    "city", // partition column name
     // partition value. Depending upon the partition column type, the
     // partition value should be created. In this example, the partition
     // column is of type StringType, so we are creating a string literal.
     Literal.ofString(city)
  );
```

--------------------------------

### Configure pyspark for Delta Lake 0.7+

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/porting.mdx

Alternatively, pass the required configurations as command line parameters when starting pyspark.

```bash
pyspark \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" \
  ...
```

--------------------------------

### Java: Resolve schema and create TransactionBuilder for append

Source: https://github.com/delta-io/delta/blob/master/kernel/USER_GUIDE.md

Use this snippet during logical plan analysis to resolve the output data schema and start transaction construction. It creates a Table object, attempts to get the latest snapshot schema, and handles TableNotFoundException for new tables by using the write operator schema. It then builds a TransactionBuilder with an engineInfo string and an Operation, and sets the schema if the table is new.

```java
import io.delta.kernel.*;
import io.delta.kernel.defaults.engine.*;

Engine myEngine = new MyEngine();
Table myTable = Table.forPath(myTablePath);

StructType writeOperatorSchema = // ... derived from the query operator tree ...
StructType dataSchema;
boolean isNewTable = false;

try {
  Snapshot mySnapshot = myTable.getLatestSnapshot(myEngine);
  dataSchema = mySnapshot.getSchema(myEngine);

  // .. check dataSchema and writeOperatorSchema match ...
} catch(TableNotFoundException e) {
  isNewTable = true;
  dataSchema = writeOperatorSchema;
}

TransactionBuilder txnBuilder =
  myTable.createTransactionBuilder(
    myEngine,
    "Examples", /* engineInfo - connector can add its own identifier which is noted in the Delta Log */ 
    Operation /* What is the operation we are trying to perform? This is noted in the Delta Log */
  );

if (isNewTable) {
  // For a new table set the table schema in the transaction builder
  txnBuilder = txnBuilder.withSchema(engine, dataSchema)
}
```

--------------------------------

### DeltaTable entry point for Python

Source: https://github.com/delta-io/delta/blob/master/examples/cheat_sheet/delta_lake_cheat_sheet.pdf

Shows how to create a DeltaTable object for programmatic interactions.

```python
from delta.tables import *
deltaTable = DeltaTable.forName(spark, tableName)
deltaTable = DeltaTable.forPath(spark, "delta.`path/to/table`")
```

--------------------------------

### Build docs for production

Source: https://github.com/delta-io/delta/blob/master/docs/README.md

Build the docs for production. The resulting files will be in docs/dist.

```python
python3 docs/generate_docs.py --api-docs
```

--------------------------------

### Scala: Read changes by version (starting only)

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-change-data-feed.mdx

Use the readChangeFeed option with only a starting version.

```scala
spark.read.format("delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("myDeltaTable")
```

--------------------------------

### Example add action for a partitioned table (JSON)

Source: https://github.com/delta-io/delta/blob/master/PROTOCOL.md

Shows a JSON add action with partitionValues populated. The dataChange flag is true, and baseRowId and defaultRowCommitVersion are included.

```json
{
  "add": {
    "path": "date=2017-12-10/part-000...c000.gz.parquet",
    "partitionValues": {"date": "2017-12-10"},
    "size": 841454,
    "modificationTime": 1512909768000,
    "dataChange": true,
    "baseRowId": 4071,
    "defaultRowCommitVersion": 41,
    "stats": "{\"numRecords\":1,\"minValues\":{\"val..."
  }
}
```

--------------------------------

### Python: Read changes by version (starting only)

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-change-data-feed.mdx

Use the readChangeFeed option with only a starting version.

```python
spark.read.format("delta") \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("myDeltaTable")
```

--------------------------------

### Start Spark Connect Server with Delta Plugins

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-spark-connect.mdx

Starts the Spark Connect server with Delta Lake Connect plugins. Requires downloading spark-4.0.0-bin-hadoop3.tgz from Spark 4.0.0. The command sets the necessary Spark SQL extensions and Delta catalog, and registers Delta relation and command plugins for Spark Connect.

```bash
sbin/start-connect-server.sh \ 
  --packages io.delta:delta-connect-server_2.13:4.0.0,com.google.protobuf:protobuf-java:3.25.1 \ 
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" \
  --conf "spark.connect.extensions.relation.classes=org.apache.spark.sql.connect.delta.DeltaRelationPlugin" \
  --conf "spark.connect.extensions.command.classes=org.apache.spark.sql.connect.delta.DeltaCommandPlugin"
```

--------------------------------

### Example: Replace columns DDL

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-batch.mdx

Example DDL that replaces columns of the boxes table.

```sql
ALTER TABLE boxes REPLACE COLUMNS (colC STRING, colB STRUCT<field2:STRING, nested:STRING, field1:STRING>, colA STRING)
```

--------------------------------

### DESCRIBE DETAIL Java example

Source: https://github.com/delta-io/delta/blob/master/docs/src/content/docs/delta-utility/index.mdx

Java code using DeltaTable.forPath and detail() to retrieve table details. Requires the io.delta.tables package.

```java
import io.delta.tables.*;

DeltaTable deltaTable = DeltaTable.forPath(spark, pathToTable);

DataFrame detailDF = deltaTable.detail();
```

--------------------------------

### First commit: create table with files 0-4

Source: https://github.com/delta-io/delta/blob/master/kernel/kernel-defaults/src/test/resources/kernel-pagination-all-jsons/info.txt

Creates the Delta table by writing 50 rows (0-49) partitioned into 5 files. This is the initial commit.

```scala
spark.range(0, 50, 1, 5).write.format("delta").save(tablePath)
```

--------------------------------

### domainMetadata action example

Source: https://github.com/delta-io/delta/blob/master/PROTOCOL.md

Example of a domainMetadata action in JSON, as it appears in the Delta protocol documentation.

```json
{
  "domainMetadata": {
    "domain": "delta.deltaTableFeatureX",
    "configuration": "{\"key1\":\"value1\"}",
    "removed": false
  }
}
```