### Full Example: Setup, Write, and Shard Read

Source: https://paimon.apache.org/docs/1.4/pypaimon/python-api

A comprehensive example demonstrating the setup of a Pypaimon catalog and table, writing data in batches, and then performing shard reads. This includes reading a specific shard and verifying shard distribution by reading all shards.

```python
import pyarrow as pa
from pypaimon import CatalogFactory, Schema

# Create catalog
catalog_options = {'warehouse': 'file:///path/to/warehouse'}
catalog = CatalogFactory.create(catalog_options)
catalog.create_database("default", False)
# Define schema
pa_schema = pa.schema([
    ('user_id', pa.int64()),
    ('item_id', pa.int64()),
    ('behavior', pa.string()),
    ('dt', pa.string()),
])

# Create table and write data
schema = Schema.from_pyarrow_schema(pa_schema, partition_keys=['dt'])
catalog.create_table('default.test_table', schema, False)
table = catalog.get_table('default.test_table')

# Write data in two batches
write_builder = table.new_batch_write_builder()

# First write
table_write = write_builder.new_write()
table_commit = write_builder.new_commit()
data1 = {
    'user_id': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14],
    'item_id': [1001, 1002, 1003, 1004, 1005, 1006, 1007, 1008, 1009, 1010, 1011, 1012, 1013, 1014],
    'behavior': ['a', 'b', 'c', None, 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm'],
    'dt': ['p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2', 'p1'],
}
pa_table = pa.Table.from_pydict(data1, schema=pa_schema)
table_write.write_arrow(pa_table)
table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()

# Second write
table_write = write_builder.new_write()
table_commit = write_builder.new_commit()
data2 = {
    'user_id': [5, 6, 7, 8, 18],
    'item_id': [1005, 1006, 1007, 1008, 1018],
    'behavior': ['e', 'f', 'g', 'h', 'z'],
    'dt': ['p2', 'p1', 'p2', 'p2', 'p1'],
}
pa_table = pa.Table.from_pydict(data2, schema=pa_schema)
table_write.write_arrow(pa_table)
table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()

# Read specific shard
read_builder = table.new_read_builder()
table_read = read_builder.new_read()

# Read shard 2 out of 3 total shards
splits = read_builder.new_scan().with_shard(2, 3).plan().splits()
shard_data = table_read.to_arrow(splits)

# Verify shard distribution by reading all shards
splits1 = read_builder.new_scan().with_shard(0, 3).plan().splits()
splits2 = read_builder.new_scan().with_shard(1, 3).plan().splits()
splits3 = read_builder.new_scan().with_shard(2, 3).plan().splits()

# Combine all shards should equal full table read
all_shards_data = pa.concat_tables([
    table_read.to_arrow(splits1),
    table_read.to_arrow(splits2),
    table_read.to_arrow(splits3),
])
full_table_data = table_read.to_arrow(read_builder.new_scan().plan().splits())

```

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

### Install PyPaimon from Source

Source: https://paimon.apache.org/docs/1.4/pypaimon/overview

Install the PyPaimon package after building the source distribution.

```bash
pip3 install dist/*.tar.gz

```

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

### Install PyPaimon using pip

Source: https://paimon.apache.org/docs/1.4/pypaimon/overview

Install the PyPaimon SDK and its dependencies using pip.

```bash
pip install pypaimon

```

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

### Example: Compact Database with S3 Catalog

Source: https://paimon.apache.org/docs/1.4/maintenance/dedicated-compaction

This example demonstrates submitting a compaction job for a specific database using an S3-compatible object storage as the warehouse. It includes necessary S3 catalog configurations.

```bash
<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-1.4.1.jar \
    compact_database \
    --warehouse s3:///path/to/warehouse \
    --including_databases test_db \
    --catalog_conf s3.endpoint=https://****.com \
    --catalog_conf s3.access-key=***** \
    --catalog_conf s3.secret-key=*****
```

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

### Example: Compact Specific Partitions with Minor Strategy

Source: https://paimon.apache.org/docs/1.4/maintenance/dedicated-compaction

An example demonstrating how to compact specific partitions of a table using the Flink Action Jar, including setting table configurations and catalog connection details.

```bash
<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-1.4.1.jar \
    compact \
    --warehouse s3:///path/to/warehouse \
    --database test_db \
    --table test_table \
    --partition dt=20221126,hh=08 \
    --partition dt=20221127,hh=09 \
    --table_conf sink.parallelism=10 \
    --compact_strategy minor \
    --catalog_conf s3.endpoint=https://****.com \
    --catalog_conf s3.access-key=***** \
    --catalog_conf s3.secret-key=*****
```

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

### Example: Compact Database in Combined Mode with S3 Catalog

Source: https://paimon.apache.org/docs/1.4/maintenance/dedicated-compaction

Submit a compaction job for a database in 'combined' mode, enabling automatic compaction of newly added tables. This example uses an S3 catalog and specifies table-level configurations.

```bash
<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-1.4.1.jar \
    compact_database \
    --warehouse s3:///path/to/warehouse \
    --including_databases test_db \
    --mode combined \
    --catalog_conf s3.endpoint=https://****.com \
    --catalog_conf s3.access-key=***** \
    --catalog_conf s3.secret-key=***** \
    --table_conf continuous.discovery-interval=*****
```

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

### Filesystem Catalog Configuration

Source: https://paimon.apache.org/docs/1.4/pypaimon/cli

Example configuration for a Paimon filesystem catalog.

```yaml
metastore: filesystem
warehouse: /path/to/warehouse

```

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

### Install Paimon Connector for Trino

Source: https://paimon.apache.org/docs/1.4/ecosystem/trino

Installs the Paimon Trino connector by extracting the plugin tarball into the Trino plugin directory. Ensure TRINO_HOME is set correctly.

```bash
tar -zxf paimon-trino-<trino-version>-1.4.1-plugin.tar.gz -C ${TRINO_HOME}/plugin

```

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

### Example: Migrate Table using Flink Action

Source: https://paimon.apache.org/docs/1.4/migration/migration-from-hive

An example demonstrating the command-line execution of the Paimon Flink Action JAR to migrate a specific Hive table. This shows concrete values for warehouse, catalog configuration, source type, and database.

```bash
<FLINK_HOME>/flink run ./paimon-flink-action-1.4.1.jar migrate_table \
--warehouse /path/to/warehouse \
--catalog_conf uri=thrift://localhost:9083 \
--catalog_conf metastore=hive \
--source_type hive \
--database default

```

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

### Get Help for compact_database Action

Source: https://paimon.apache.org/docs/1.4/maintenance/dedicated-compaction

Run this command to display the help message for the `compact_database` action, showing all available options and their usage.

```bash
<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-1.4.1.jar \
    compact_database --help
```

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

### Specify Paimon JAR using --packages

Source: https://paimon.apache.org/docs/1.4/spark/quick-start

Starts spark-sql with the Paimon JAR included via the --packages argument, using Maven coordinates. Ensure the version and Scala compatibility match your setup.

```bash
spark-sql ... --packages org.apache.paimon:paimon-spark-3.5_2.12:1.4.1

```

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

### HTTP Response Body Example

Source: https://paimon.apache.org/docs/1.4/flink/sql-write

Example of a successful response body from the HTTP server for the http-report partition mark done action.

```json
{
    "result": "success"
}
```

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

### Get Tag Response Sample

Source: https://paimon.apache.org/docs/1.4/concepts/rest/rest-api

Example JSON response when retrieving a specific tag. Includes tag details, snapshot information, creation time, and retention period.

```json
{
  "tagName": "string",
  "snapshot": {
    "version": 0,
    "id": 0,
    "schemaId": 0,
    "baseManifestList": "string",
    "deltaManifestList": "string",
    "changelogManifestList": "string",
    "indexManifest": "string",
    "commitUser": "string",
    "commitIdentifier": "string",
    "commitKind": "APPEND",
    "timeMillis": 0,
    "logOffsets": {
      "property1": 0,
      "property2": 0
    },
    "totalRecordCount": 0,
    "deltaRecordCount": 0,
    "changelogRecordCount": 0,
    "watermark": 0,
    "statistics": "string"
  },
  "tagCreateTime": 0,
  "tagTimeRetained": "string"
}
```

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

### Start Flink Local Cluster

Source: https://paimon.apache.org/docs/1.4/flink/quick-start

Initiate a local Flink cluster using the provided start script.

```bash
<FLINK_HOME>/bin/start-cluster.sh

```

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

### Get View Response Sample

Source: https://paimon.apache.org/docs/1.4/concepts/rest/rest-api

This is a sample JSON response for a successful GET request to retrieve view details. It includes schema information, owner, and creation/update metadata.

```json
{
  "id": "string",
  "name": "string",
  "schema": {
    "fields": [
      {
        "id": 0,
        "name": "string",
        "type": "string",
        "description": "string"
}
],
    "query": "string",
    "dialects": {
      "property1": "string",
      "property2": "string"
},
    "comment": "string",
    "options": {
      "property1": "string",
      "property2": "string"
}
},
  "owner": "string",
  "createdAt": null,
  "createdBy": "string",
  "updatedAt": null,
  "updatedBy": "string"
}
```

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

### Aggregate Pushdown Example

Source: https://paimon.apache.org/docs/1.4/append-table/overview

Demonstrates a simple COUNT aggregate query that can be accelerated by Paimon's pushdown capabilities.

```sql
SELECT COUNT(*) FROM TABLE WHERE DT = '20230101';
```

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

### Start Flink SQL Client

Source: https://paimon.apache.org/docs/1.4/flink/quick-start

Launch the Flink SQL client to execute SQL scripts.

```bash
<FLINK_HOME>/bin/sql-client.sh

```

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

### Example Flink Action Compact Table

Source: https://paimon.apache.org/docs/1.4/learn-paimon/understand-files

An example Flink Action command to compact a Paimon table located at a specific file path.

```bash
./bin/flink run \
    ./lib/paimon-flink-action-1.4.1.jar \
    compact \
    --path file:///tmp/paimon/default.db/T
```

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

### HTTP Post Request Body Example

Source: https://paimon.apache.org/docs/1.4/flink/sql-write

Example of the JSON payload sent to the HTTP server when using the http-report partition mark done action.

```json
{
    "table": "table fullName",
    "path": "table location path",
    "partition": "mark done partition",
    "params" : "custom params"
}
```

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

### Start Spark SQL with Paimon S3 Catalog

Source: https://paimon.apache.org/docs/1.4/maintenance/filesystems

Configure and start Spark SQL to use a Paimon catalog with S3. Place the necessary Paimon JARs in Spark's jars directory.

```bash
spark-sql \
  --conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions \
  --conf spark.sql.catalog.paimon=org.apache.paimon.spark.SparkCatalog \
  --conf spark.sql.catalog.paimon.warehouse=s3://<bucket>/<path> \
  --conf spark.sql.catalog.paimon.s3.endpoint=your-endpoint-hostname \
  --conf spark.sql.catalog.paimon.s3.access-key=xxx \
  --conf spark.sql.catalog.paimon.s3.secret-key=yyy
```

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

### Example: Run Incremental Clustering with Flink Action

Source: https://paimon.apache.org/docs/1.4/append-table/incremental-clustering

An example of running an incremental clustering job using Flink Action with specific configurations for warehouse, database, table, sink parallelism, compact strategy, and catalog settings.

```bash
<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-1.4.1.jar \
    compact \
    --warehouse s3:///path/to/warehouse \
    --database test_db \
    --table test_table \
    --table_conf sink.parallelism=2 \
    --compact_strategy minor \
    --catalog_conf s3.endpoint=https://****.com \
    --catalog_conf s3.access-key=***** \
    --catalog_conf s3.secret-key=*****
```

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

### Spark SQL Configuration for Paimon and Iceberg

Source: https://paimon.apache.org/docs/1.4/iceberg/primary-key-table

Starts the spark-sql client with necessary JARs and configurations for Paimon and Iceberg catalogs, including extensions.

```bash
spark-sql --jars <path-to-paimon-jar> \
    --conf spark.sql.catalog.paimon_catalog=org.apache.paimon.spark.SparkCatalog \
    --conf spark.sql.catalog.paimon_catalog.warehouse=<path-to-warehouse> \
    --packages org.apache.iceberg:iceberg-spark-runtime-<iceberg-version> \
    --conf spark.sql.catalog.iceberg_catalog=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.iceberg_catalog.type=hadoop \
    --conf spark.sql.catalog.iceberg_catalog.warehouse=<path-to-warehouse>/iceberg \
    --conf spark.sql.catalog.iceberg_catalog.cache-enabled=false \# disable iceberg catalog caching to quickly see the result \
    --conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions,org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
```

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

### REST Catalog Configuration

Source: https://paimon.apache.org/docs/1.4/pypaimon/cli

Example configuration for a Paimon REST catalog.

```yaml
metastore: rest
uri: http://localhost:8080
warehouse: catalog_name

```

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

### Read Paimon Table with Options

Source: https://paimon.apache.org/docs/1.4/pypaimon/cli

Examples of reading Paimon table data with various options like limit, select, where, and format.

```bash
# Read with limit
paimon table read mydb.users -l 50

# Read specific columns
paimon table read mydb.users -s id,name,age

# Filter with WHERE clause
paimon table read mydb.users --where "age > 18"

# Combine select, where, and limit
paimon table read mydb.users -s id,name -w "age >= 20 AND city = 'Beijing'" -l 50

# Output as JSON (for programmatic use)
paimon table read mydb.users --format json

```

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

### Create Table with Partitioning and Bucketing

Source: https://paimon.apache.org/docs/1.4/learn-paimon/understand-files

Example of creating a Paimon table with specified partitioning and bucket count. Adjusting the bucket count can help manage the number of small files.

```sql
CREATE TABLE MyTable (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
) PARTITIONED BY (dt, hh) WITH (
    'bucket' = '10'
);

```

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

### Install PyJindoSDK

Source: https://paimon.apache.org/docs/1.4/pypaimon/pyjindosdk-support

Install the pyjindosdk package using pip. PyPaimon will automatically use it for OSS access after installation.

```bash
pip install pyjindosdk

```

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

### Prepare and Read Specific Shards

Source: https://paimon.apache.org/docs/1.4/pypaimon/python-api

Demonstrates how to prepare a read builder and read specific shards of a table. It shows how to read a single shard and how to combine results from multiple shards.

```python
# Prepare read builder
table = catalog.get_table('database_name.table_name')
read_builder = table.new_read_builder()
table_read = read_builder.new_read()

# Read the second shard (index 1) out of 3 total shards
splits = read_builder.new_scan().with_shard(1, 3).plan().splits()

# Read all shards and concatenate results
splits1 = read_builder.new_scan().with_shard(0, 3).plan().splits()
splits2 = read_builder.new_scan().with_shard(1, 3).plan().splits()
splits3 = read_builder.new_scan().with_shard(2, 3).plan().splits()

# Combine results from all shards

all_splits = splits1 + splits2 + splits3
pa_table = table_read.to_arrow(all_splits)
```

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

### Expire Changelogs Example (Flink 1.18)

Source: https://paimon.apache.org/docs/1.4/flink/procedures

Examples of expiring changelogs using the Flink 1.18 syntax with indexed arguments.

```sql
-- for Flink 1.18
CALL sys.expire_changelogs('default.T', 4, 2, '2024-01-01 12:00:00', 2)
CALL sys.expire_changelogs('default.T', true)

```

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

### Expire Snapshots Example (Flink 1.18)

Source: https://paimon.apache.org/docs/1.4/flink/procedures

An example of expiring snapshots using the Flink 1.18 syntax with indexed arguments.

```sql
-- for Flink 1.18
CALL sys.expire_snapshots('default.T', 2)

```

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

### Start Spark SQL with Paimon OBS Catalog

Source: https://paimon.apache.org/docs/1.4/maintenance/filesystems

Launch Spark SQL with Paimon extensions and configure the OBS catalog. Place Paimon OBS and Spark connector JARs in Spark's jars directory.

```bash
spark-sql \
  --conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions \
  --conf spark.sql.catalog.paimon=org.apache.paimon.spark.SparkCatalog \
  --conf spark.sql.catalog.paimon.warehouse=obs://<bucket>/<path> \
  --conf spark.sql.catalog.paimon.fs.obs.endpoint=obs-endpoint-hostname \
  --conf spark.sql.catalog.paimon.fs.obs.access.key=xxx \
  --conf spark.sql.catalog.paimon.fs.obs.secret.key=yyy
```

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

### Create Filesystem Catalog

Source: https://paimon.apache.org/docs/1.4/pypaimon/python-api

Demonstrates how to create a catalog that uses the local filesystem as its backend. Catalog options, including the warehouse path, are provided as a dictionary.

```APIDOC
## Create Catalog (Filesystem)

### Description
Creates a catalog using the local filesystem. This is the first step before interacting with tables.

### Method
`CatalogFactory.create(catalog_options)`

### Parameters
#### Catalog Options
- **warehouse** (string) - Required - The path to the warehouse directory on the local filesystem.

### Request Example
```python
from pypaimon import CatalogFactory

catalog_options = {
    'warehouse': 'file:///path/to/warehouse'
}
catalog = CatalogFactory.create(catalog_options)
```
```

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

### Create Table Like

Source: https://paimon.apache.org/docs/1.4/flink/sql-ddl

Creates a new table with the same schema, partition, and table properties as an existing table. Use EXCLUDING OPTIONS to omit table properties.

```sql
CREATE TABLE my_table (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING,
    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
);

CREATE TABLE my_table_like LIKE my_table (EXCLUDING OPTIONS);
```

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

### Create Table with Multiple Partition Fields and values-time Strategy

Source: https://paimon.apache.org/docs/1.4/maintenance/manage-partitions

This example shows how to create a table partitioned by multiple fields ('other_key', 'dt') and configure partition expiration using the 'values-time' strategy. It includes expiration time, check interval, timestamp formatter, and a pattern to extract the timestamp from the 'dt' partition field.

```sql
CREATE TABLE t (...) PARTITIONED BY (other_key, dt) WITH (
    'partition.expiration-time' = '7 d',
    'partition.expiration-check-interval' = '1 d',
    'partition.timestamp-formatter' = 'yyyyMMdd',
    'partition.timestamp-pattern' = '$dt'
);
```

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

### Debezium BSON Event Message Example

Source: https://paimon.apache.org/docs/1.4/cdc-ingestion/kafka-cdc

An example of an update operation captured from a MongoDB customers collection in JSON format, illustrating the structure of Debezium events.

```json
{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "string",
        "optional": true,
        "name": "io.debezium.data.Json",
        "version": 1,
        "field": "before"
      },
      {
        "type": "string",
        "optional": true,
        "name": "io.debezium.data.Json",
        "version": 1,
        "field": "after"
      },
      ...
    ]
  },
  "payload": {
    "before": "{\"_id\": {\"$oid\" : \"596e275826f08b2730779e1f\"}, \"name\" : \"Anne\", \"create_time\" : {\"$numberLong\" : \"1558965506000\"}, \"tags\":[\"success\"]}",
    "after": "{\"_id\": {\"$oid\" : \"596e275826f08b2730779e1f\"}, \"name\" : \"Anne\", \"create_time\" : {\"$numberLong\" : \"1558965506000\"}, \"tags\":[\"passion\",\"success\"]}",
    "source": {
      "db": "inventory",
      "rs": "rs0",
      "collection": "customers",
      ...
    },
    "op": "u",
    "ts_ms": 1558965515240,
    "ts_us": 1558965515240142,
    "ts_ns": 1558965515240142879
  }
}

```

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

### Create Table

Source: https://paimon.apache.org/docs/1.4/pypaimon/python-api

Details the process of creating a table, including defining its schema, partition keys, primary keys, and options. It also shows how to retrieve an existing table.

```APIDOC
## Create Table

### Description
Creates a table with a defined schema, partition keys, primary keys, and options. Also demonstrates how to retrieve a table.

### Method
`catalog.create_table(identifier, schema, ignore_if_exists)`
`catalog.get_table(identifier)`

### Parameters for `create_table`
- **identifier** (string) - Required - The fully qualified name of the table (e.g., 'database_name.table_name').
- **schema** (Schema) - Required - The Paimon Schema object defining the table structure.
- **ignore_if_exists** (boolean) - Optional - If True, does not raise an error if the table exists. Defaults to True.

### Parameters for `get_table`
- **identifier** (string) - Required - The fully qualified name of the table (e.g., 'database_name.table_name').

### Schema Definition
#### From PyArrow Schema
```python
import pyarrow as pa
from pypaimon import Schema

pa_schema = pa.schema([
    ('dt', pa.string()),
    ('hh', pa.string()),
    ('pk', pa.int64()),
    ('value', pa.string())
])

schema = Schema.from_pyarrow_schema(
    pa_schema=pa_schema,
    partition_keys=['dt', 'hh'],
    primary_keys=['dt', 'hh', 'pk'],
    options={'bucket': '2'},
    comment='my test table'
)
```
#### From Pandas DataFrame
```python
import pandas as pd
import pyarrow as pa
from pypaimon import Schema

data = {
    'dt': ['2024-01-01', '2024-01-01', '2024-01-02'],
    'hh': ['12', '15', '20'],
    'pk': [1, 2, 3],
    'value': ['a', 'b', 'c'],
}
dataframe = pd.DataFrame(data)
record_batch = pa.RecordBatch.from_pandas(dataframe)
schema = Schema.from_pyarrow_schema(
    pa_schema=record_batch.schema,
    partition_keys=['dt', 'hh'],
    primary_keys=['dt', 'hh', 'pk'],
    options={'bucket': '2'},
    comment='my test table'
)
```

### Request Example (Create Table)
```python
schema = ... # Obtained from above methods
catalog.create_table(
    identifier='database_name.table_name',
    schema=schema,
    ignore_if_exists=True
)
```

### Request Example (Get Table)
```python
table = catalog.get_table('database_name.table_name')
```
```

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

### Expire Changelogs Example (Flink 1.19+ Named Arguments)

Source: https://paimon.apache.org/docs/1.4/flink/procedures

Examples of expiring changelogs using named arguments for Flink 1.19 and later, demonstrating various options.

```sql
-- for Flink 1.19 and later
CALL sys.expire_changelogs(`table` => 'default.T', retain_max => 2)
CALL sys.expire_changelogs(`table` => 'default.T', older_than => '2024-01-01 12:00:00')
CALL sys.expire_changelogs(`table` => 'default.T', older_than => '2024-01-01 12:00:00', retain_min => 10)
CALL sys.expire_changelogs(`table` => 'default.T', older_than => '2024-01-01 12:00:00', max_deletes => 10)
CALL sys.expire_changelogs(`table` => 'default.T', delete_all => true)

```

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

### Expire Snapshots Example (Flink 1.19+ Named Arguments)

Source: https://paimon.apache.org/docs/1.4/flink/procedures

Examples of expiring snapshots using named arguments for Flink 1.19 and later, demonstrating various options.

```sql
-- for Flink 1.19 and later
CALL sys.expire_snapshots(`table` => 'default.T', retain_max => 2)
CALL sys.expire_snapshots(`table` => 'default.T', older_than => '2024-01-01 12:00:00')
CALL sys.expire_snapshots(`table` => 'default.T', older_than => '2024-01-01 12:00:00', retain_min => 10)
CALL sys.expire_snapshots(`table` => 'default.T', older_than => '2024-01-01 12:00:00', max_deletes => 10, options => 'snapshot.expire.limit=1')

```

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

### Create Table As Select (Partitioned)

Source: https://paimon.apache.org/docs/1.4/spark/sql-ddl

Creates a new partitioned table and populates it with query results. Partitions can be specified for the new table.

```sql
/* partitioned table*/
CREATE TABLE my_table_partition (
      user_id BIGINT,
      item_id BIGINT,
      behavior STRING,
      dt STRING,
      hh STRING
) PARTITIONED BY (dt, hh);
CREATE TABLE my_table_partition_as PARTITIONED BY (dt) AS SELECT * FROM my_table_partition;

```

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

### Add Paimon JAR to Hive (auxlib)

Source: https://paimon.apache.org/docs/1.4/ecosystem/hive

Installs the Paimon Hive connector JAR by placing it in the 'auxlib' folder of the Hive installation. This is a recommended method for enabling Paimon support.

```bash
cp paimon-hive-connector-1.4.1.jar auxlib/
```

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

### Create Table As Select (Table Properties)

Source: https://paimon.apache.org/docs/1.4/spark/sql-ddl

Creates a new table with specified table properties (e.g., file format) and populates it with query results.

```sql
/* change TBLPROPERTIES */
CREATE TABLE my_table_options (
       user_id BIGINT,
       item_id BIGINT
) TBLPROPERTIES ('file.format' = 'orc');
CREATE TABLE my_table_options_as TBLPROPERTIES ('file.format' = 'parquet') AS SELECT * FROM my_table_options;

```

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

### Start Spark SQL with Paimon OSS Catalog

Source: https://paimon.apache.org/docs/1.4/maintenance/filesystems

Configure and launch Spark SQL to use Paimon with OSS as the warehouse. Place the required Paimon JARs in Spark's jars directory.

```bash
spark-sql \
  --conf spark.sql.extensions=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions \
  --conf spark.sql.catalog.paimon=org.apache.paimon.spark.SparkCatalog \
  --conf spark.sql.catalog.paimon.warehouse=oss://<bucket>/<path> \
  --conf spark.sql.catalog.paimon.fs.oss.endpoint=oss-cn-hangzhou.aliyuncs.com \
  --conf spark.sql.catalog.paimon.fs.oss.accessKeyId=xxx \
  --conf spark.sql.catalog.paimon.fs.oss.accessKeySecret=yyy

```

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

### Build PyPaimon Source Package

Source: https://paimon.apache.org/docs/1.4/pypaimon/overview

Build the source distribution package for PyPaimon using the setup.py script.

```bash
python3 setup.py sdist

```

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

### Stream Write and Commit Example

Source: https://paimon.apache.org/docs/1.4/api/java-api.html

Demonstrates the complete lifecycle of writing records in a streaming fashion and committing them. Ensure commitIdentifier is incremented for each new commit batch. This example is suitable for distributed processing where records are written and then committed.

```java
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.sink.StreamWriteBuilder;

import java.util.List;

public class StreamWriteTable {

    public static void main(String[] args) throws Exception {
        // 1. Create a WriteBuilder (Serializable)
        Table table = GetTable.getTable();
        StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder();

        // 2. Write records in distributed tasks
        StreamTableWrite write = writeBuilder.newWrite();
        // commitIdentifier like Flink checkpointId
        long commitIdentifier = 0;

        while (true) {
            GenericRow record1 = GenericRow.of(BinaryString.fromString("Alice"), 12);
            GenericRow record2 = GenericRow.of(BinaryString.fromString("Bob"), 5);
            GenericRow record3 = GenericRow.of(BinaryString.fromString("Emily"), 18);

            // If this is a distributed write, you can use writeBuilder.newWriteSelector.
            // WriteSelector determines to which logical downstream writers a record should be written to.
            // If it returns empty, no data distribution is required.

            write.write(record1);
            write.write(record2);
            write.write(record3);
            List<CommitMessage> messages = write.prepareCommit(false, commitIdentifier);
            commitIdentifier++;

            // 3. Collect all CommitMessages to a global node and commit
            StreamTableCommit commit = writeBuilder.newCommit();
            commit.commit(commitIdentifier, messages);

            // 4. When failure occurs and you're not sure if the commit process is successful,
            //    you can use `filterAndCommit` to retry the commit process.
            //    Succeeded commits will be automatically skipped.
            /*
            Map<Long, List<CommitMessage>> commitIdentifiersAndMessages = new HashMap<>();
            commitIdentifiersAndMessages.put(commitIdentifier, messages);
            commit.filterAndCommit(commitIdentifiersAndMessages);
            */

            Thread.sleep(1000);
        }
    }
}

```

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

### Create Table with update-time Expiration Strategy

Source: https://paimon.apache.org/docs/1.4/maintenance/manage-partitions

This example demonstrates creating a partitioned table using the 'update-time' expiration strategy. It sets the expiration time and check interval, and uses 'update-time' to expire partitions based on their last update time. This is useful when partition values are not date-formatted.

```sql
CREATE TABLE t (...) PARTITIONED BY (dt) WITH (
    'partition.expiration-time' = '7 d',
    'partition.expiration-check-interval' = '1 d',
    'partition.expiration-strategy' = 'update-time'
);

-- The last update time of the partition is now, so it will not expire.
insert into t values('pk', '2024-01-01');
-- Support non-date formatted partition.
insert into t values('pk', 'par-1'); 



```

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

### Create Paimon Filesystem Catalog

Source: https://paimon.apache.org/docs/1.4/program-api/cpp-api

Instantiate a Paimon Catalog using a root path and options. Currently, only filesystem catalogs are supported. Ensure keys and values in the options map are strings.

```cpp
#include "paimon/catalog/catalog.h"

// Note that keys and values are all string
std::map<std::string, std::string> options;
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::Catalog> catalog,
                       paimon::Catalog::Create(root_path, options));
```

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

### Get function

Source: https://paimon.apache.org/docs/1.4/concepts/rest/rest-api

Retrieves a specific function by its name and database.

```APIDOC
## Get function

### Description
Retrieves a specific function by its name and database.

### Method
GET

### Endpoint
/v1/{prefix}/databases/{database}/functions/{function}

### Parameters
#### Path Parameters
- **prefix** (string) - Required - 
- **database** (string) - Required - 
- **function** (string) - Required - 

### Response
#### Success Response (200)
OK
- **name** (string) - 
- **inputParams** (array) - 
  - **id** (integer) - 
  - **name** (string) - 
  - **type** (string) - 
  - **description** (string) - 
- **returnParams** (array) - 
  - **id** (integer) - 
  - **name** (string) - 
  - **type** (string) - 
  - **description** (string) - 
- **deterministic** (boolean) - 
- **definitions** (object) - 
  - **property1** (object) - 
    - **type** (string) - 
    - **fileResources** (array) - 
      - **resourceType** (string) - 
      - **uri** (string) - 
    - **language** (string) - 
    - **className** (string) - 
    - **functionName** (string) - 
  - **property2** (object) - 
    - **type** (string) - 
    - **fileResources** (array) - 
      - **resourceType** (string) - 
      - **uri** (string) - 
    - **language** (string) - 
    - **className** (string) - 
    - **functionName** (string) - 
- **comment** (string) - 
- **options** (object) - 
  - **property1** (string) - 
  - **property2** (string) - 
- **owner** (string) - 
- **createdAt** (null) - 
- **createdBy** (string) - 
- **updatedAt** (null) - 
- **updatedBy** (string) - 

#### Response Example
```json
{
  "name": "string",
  "inputParams": [
    {
      "id": 0,
      "name": "string",
      "type": "string",
      "description": "string"
    }
  ],
  "returnParams": [
    {
      "id": 0,
      "name": "string",
      "type": "string",
      "description": "string"
    }
  ],
  "deterministic": true,
  "definitions": {
    "property1": {
      "type": "string",
      "fileResources": [
        {
          "resourceType": "string",
          "uri": "string"
        }
      ],
      "language": "string",
      "className": "string",
      "functionName": "string"
    },
    "property2": {
      "type": "string",
      "fileResources": [
        {
          "resourceType": "string",
          "uri": "string"
        }
      ],
      "language": "string",
      "className": "string",
      "functionName": "string"
    }
  },
  "comment": "string",
  "options": {
    "property1": "string",
    "property2": "string"
  },
  "owner": "string",
  "createdAt": null,
  "createdBy": "string",
  "updatedAt": null,
  "updatedBy": "string"
}
```
```

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

### Create Paimon Catalog and Table

Source: https://paimon.apache.org/docs/1.4/flink/sql-lookup

Sets up a Paimon catalog and a 'customers' table. This is a prerequisite for performing lookup joins.

```sql
CREATE CATALOG my_catalog WITH (
  'type'='paimon',
  'warehouse'='hdfs://nn:8020/warehouse/path' -- or 'file://tmp/foo/bar'
);

USE CATALOG my_catalog;

-- Create a table in paimon catalog
CREATE TABLE customers (
    id INT PRIMARY KEY NOT ENFORCED,
    name STRING,
    country STRING,
    zip STRING
);
```

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

### Get Database

Source: https://paimon.apache.org/docs/1.4/concepts/rest/rest-api

Retrieves details of a specific database by its name.

```APIDOC
## GET /v1/{prefix}/databases/{database}

### Description
Get a database by database name.

### Method
GET

### Endpoint
/v1/{prefix}/databases/{database}

### Parameters
#### Path Parameters
- **prefix** (string) - Required
- **database** (string) - Required

### Response
#### Success Response (200)
Get a database by database name.

#### Response Example
```json
{
  "id": "string",
  "name": "string",
  "location": "string",
  "options": {
    "property1": "string",
    "property2": "string"
  },
  "owner": "string",
  "createdAt": null,
  "createdBy": "string",
  "updatedAt": null,
  "updatedBy": "string"
}
```
```

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

### Show Create Table Statement

Source: https://paimon.apache.org/docs/1.4/spark/auxiliary

SHOW CREATE TABLE returns the exact CREATE TABLE or CREATE VIEW statement used to create the specified table or view.

```sql
SHOW CREATE TABLE my_table;

```

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

### List Databases using Catalog API

Source: https://paimon.apache.org/docs/1.4/program-api/catalog-api

Provides an example of how to retrieve a list of all available databases managed by the Paimon Catalog API. Requires importing java.util.List.

```java
import org.apache.paimon.catalog.Catalog;

import java.util.List;

public class ListDatabases {

    public static void main(String[] args) {
        Catalog catalog = CreateCatalog.createFilesystemCatalog();
        List<String> databases = catalog.listDatabases();
    }
}

```

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

### Rescaling Partitions with Different Bucket Numbers

Source: https://paimon.apache.org/docs/1.4/maintenance/rescale-bucket

Demonstrates how to set different bucket numbers for different partitions of a table and then overwrite those partitions individually.

```sql
ALTER TABLE my_table SET ('bucket' = '4');
INSERT OVERWRITE my_table PARTITION (dt = '2022-01-01')
SELECT * FROM ...;
  
ALTER TABLE my_table SET ('bucket' = '8');
INSERT OVERWRITE my_table PARTITION (dt = '2022-01-02')
SELECT * FROM ...;

```

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

### Get tag

Source: https://paimon.apache.org/docs/1.4/concepts/rest/rest-api

Retrieves details of a specific tag associated with a table.

```APIDOC
## GET /v1/{prefix}/databases/{database}/tables/{table}/tags/{tag}

### Description
Retrieves details of a specific tag associated with a table.

### Method
GET

### Endpoint
/v1/{prefix}/databases/{database}/tables/{table}/tags/{tag}

### Parameters
#### Path Parameters
- **prefix** (string) - Required
- **database** (string) - Required
- **table** (string) - Required
- **tag** (string) - Required

### Response
#### Success Response (200)
OK

#### Response Example
```json
{
  "tagName": "string",
  "snapshot": {
    "version": 0,
    "id": 0,
    "schemaId": 0,
    "baseManifestList": "string",
    "deltaManifestList": "string",
    "changelogManifestList": "string",
    "indexManifest": "string",
    "commitUser": "string",
    "commitIdentifier": "string",
    "commitKind": "APPEND",
    "timeMillis": 0,
    "logOffsets": {
      "property1": 0,
      "property2": 0
    },
    "totalRecordCount": 0,
    "deltaRecordCount": 0,
    "changelogRecordCount": 0,
    "watermark": 0,
    "statistics": "string"
  },
  "tagCreateTime": 0,
  "tagTimeRetained": "string"
}
```
```

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

### Get table by id

Source: https://paimon.apache.org/docs/1.4/concepts/rest/rest-api

Retrieves a specific table by its unique ID.

```APIDOC
## GET /v1/{prefix}/tables/id/{tableId}

### Description
Retrieves a specific table by its unique ID.

### Method
GET

### Endpoint
/v1/{prefix}/tables/id/{tableId}

### Parameters
#### Path Parameters
- **prefix** (string) - Required - 
- **tableId** (string) - Required - 

### Response
#### Success Response (200)
OK
- **id** (string) - 
- **database** (string) - 
- **name** (string) - 
- **path** (string) - 
- **isExternal** (boolean) - 
- **schemaId** (integer) - 
- **schema** (object) - 
  - **fields** (array) - 
    - **id** (integer) - 
    - **name** (string) - 
    - **type** (string) - 
    - **description** (string) - 
  - **partitionKeys** (array) - 
    - (string) - 
  - **primaryKeys** (array) - 
    - (string) - 
  - **options** (object) - 
    - **property1** (string) - 
    - **property2** (string) - 
  - **comment** (string) - 
- **owner** (string) - 
- **createdAt** (null) - 
- **createdBy** (string) - 
- **updatedAt** (null) - 
- **updatedBy** (string) - 

#### Response Example
```json
{
  "id": "string",
  "database": "string",
  "name": "string",
  "path": "string",
  "isExternal": true,
  "schemaId": 0,
  "schema": {
    "fields": [
      {
        "id": 0,
        "name": "string",
        "type": "string",
        "description": "string"
      }
    ],
    "partitionKeys": [
      "string"
    ],
    "primaryKeys": [
      "string"
    ],
    "options": {
      "property1": "string",
      "property2": "string"
    },
    "comment": "string"
  },
  "owner": "string",
  "createdAt": null,
  "createdBy": "string",
  "updatedAt": null,
  "updatedBy": "string"
}
```

#### Error Response
- **401** - Used for 401 errors.
- **404** - Not Found - TableNotExistException, the table does not exist
- **500** - Used for server 5xx errors.
```

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

### Example Data File Structure for Primary Key Table

Source: https://paimon.apache.org/docs/1.4/concepts/spec/datafile

Illustrates the column structure of a data file for a Paimon table with a primary key. Includes key columns, value kind, sequence number, and value columns.

```sql
CREATE TABLE T (
    a INT PRIMARY KEY NOT ENFORCED,
    b INT,
    c INT
);

```

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

### Create Table As Select (Primary Key + Partition)

Source: https://paimon.apache.org/docs/1.4/spark/sql-ddl

Creates a new table with both partitioning and primary key constraints, populated from a query result. Specific partition and primary key definitions can be applied to the new table.

```sql
/* primary key + partition */
CREATE TABLE my_table_all (
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING,
    dt STRING,
    hh STRING
) PARTITIONED BY (dt, hh) TBLPROPERTIES (
    'primary-key' = 'dt,hh,user_id'
);
CREATE TABLE my_table_all_as PARTITIONED BY (dt) TBLPROPERTIES ('primary-key' = 'dt,hh') AS SELECT * FROM my_table_all;

```

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

### Get Config

Source: https://paimon.apache.org/docs/1.4/concepts/rest/rest-api

Retrieves the current configuration settings for the catalog service.

```APIDOC
## GET /v1/config

### Description
Retrieves the configuration defined in the server.

### Method
GET

### Endpoint
/v1/config

### Parameters
#### Query Parameters
- **warehouse** (string) - Required - Warehouse location or identifier to request from the service

### Response
#### Success Response (200)
Config defined in the server

#### Response Example
```json
{
  "overrides": {
    "key": "value"
  },
  "defaults": {
    "prefix": "prefix"
  }
}
```
```

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

### Alter Function

Source: https://paimon.apache.org/docs/1.4/spark/procedures

Alters an existing function, for example, by adding a new definition.

```sql
CALL sys.alter_function(`function` => 'function_identifier',
`change` => '{"action" : "addDefinition", "name" : "spark", "definition" : {"type" : "lambda", "definition" : "(Integer length, Integer width) -> { return (long) length * width; }", "language": "JAVA" } }'
)
```

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

### Dynamic vs. Static Overwrite

Source: https://paimon.apache.org/docs/1.4/flink/sql-write

Demonstrates dynamic partition overwrite (default) and static overwrite for partitioned tables. Static overwrite can be configured using OPTIONS('dynamic-partition-overwrite' = 'false').

```sql
-- Dynamic overwrite
INSERT OVERWRITE my_table SELECT ...

-- Static overwrite (Overwrite whole table)
INSERT OVERWRITE my_table /*+ OPTIONS('dynamic-partition-overwrite' = 'false') */ SELECT ...

```