This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new db25b7cd0f [docs] Restructure Program API guides and add SVG diagrams
(#9730)
db25b7cd0f is described below
commit db25b7cd0fa14789ae9a281a5717352828d3c423
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Sep 11 13:01:24 2026 +0800
[docs] Restructure Program API guides and add SVG diagrams (#9730)
---
docs/docs/program-api/catalog-api.md | 388 +++++++++----------------
docs/docs/program-api/cpp-api.md | 102 ++++---
docs/docs/program-api/file-cache.mdx | 86 ++++--
docs/docs/program-api/flink-api.mdx | 175 +++++++-----
docs/docs/program-api/index.md | 43 +++
docs/docs/program-api/java-api.mdx | 426 +++++-----------------------
docs/docs/program-api/java-reading.md | 205 +++++++++++++
docs/docs/program-api/java-types.md | 96 +++++++
docs/docs/program-api/java-writing.md | 182 ++++++++++++
docs/docs/program-api/rest-api.mdx | 37 ++-
docs/docs/program-api/rust-api.md | 25 +-
docs/sidebars.js | 25 +-
docs/static/img/program-api-local-cache.svg | 54 ++++
docs/static/img/program-api-overview.svg | 63 ++++
docs/static/img/program-api-read-flow.svg | 52 ++++
docs/static/img/program-api-write-flow.svg | 56 ++++
16 files changed, 1252 insertions(+), 763 deletions(-)
diff --git a/docs/docs/program-api/catalog-api.md
b/docs/docs/program-api/catalog-api.md
index 032f5ad9bd..3afb609413 100644
--- a/docs/docs/program-api/catalog-api.md
+++ b/docs/docs/program-api/catalog-api.md
@@ -1,6 +1,6 @@
---
title: "Catalog API"
-sidebar_position: 4
+sidebar_position: 2
---
<!--
@@ -24,314 +24,208 @@ under the License.
# Catalog API
-## Create Database
+Use `Catalog` to manage databases and tables, and to load a `Table` for
reading or writing.
+Catalog configuration and table options have different scopes: configure the
warehouse and metastore
+on the catalog; put storage and read/write behavior in the table schema's
options.
-You can use the catalog to create databases. The created databases are
persistence in the file system.
+## Set up the examples
+
+Add the [Java dependency](java-api#dependency) and save the
+[`CreateCatalog` helper](java-api#create-catalog). The snippets below are
method-body fragments.
+Place the operations you want to run inside this catalog scope:
```java
import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.catalog.PropertyChange;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
-public class CreateDatabase {
+import java.util.Arrays;
+import java.util.List;
+
+public class CatalogExample {
- public static void main(String[] args) {
- try {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- catalog.createDatabase("my_db", false);
- } catch (Catalog.DatabaseAlreadyExistException e) {
- // do something
+ public static void main(String[] args) throws Exception {
+ try (Catalog catalog = CreateCatalog.createFilesystemCatalog()) {
+ // Insert the relevant snippets here.
}
}
}
```
-## Determine Whether Database Exists
+The `ignoreIfExists` and `ignoreIfNotExists` flags handle an existing or
missing object, respectively.
+They do not suppress other validation errors. Use `false` when an unexpected
catalog state should
+fail the operation. The examples propagate exceptions; applications can handle
the corresponding
+`Catalog.*Exception` at their error-handling boundary.
-You can use the catalog to determine whether the database exists
+## Manage databases
+
+### Create Database
```java
-import org.apache.paimon.catalog.Catalog;
+catalog.createDatabase("my_db", false);
+```
-public class DatabaseExists {
+### Determine Whether Database Exists
- public static void main(String[] args) {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- boolean exists = catalog.databaseExists("my_db");
- }
-}
+```java
+boolean exists = catalog.databaseExists("my_db");
```
-## List Databases
-
-You can use the catalog to list databases.
+### List Databases
```java
-import org.apache.paimon.catalog.Catalog;
+List<String> databases = catalog.listDatabases();
+```
-import java.util.List;
+### Alter Database
-public class ListDatabases {
+Use `PropertyChange` for database properties. Hive, JDBC, and REST catalogs
support this operation;
+the filesystem catalog does not. Run this fragment with a suitable catalog,
such as one created
+with `CreateCatalog.createHiveCatalog()`, and an existing database.
- public static void main(String[] args) {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- List<String> databases = catalog.listDatabases();
- }
-}
+```java
+List<PropertyChange> changes = Arrays.asList(
+ PropertyChange.setProperty("owner", "analytics"),
+ PropertyChange.removeProperty("obsolete-property"));
+catalog.alterDatabase("my_db", changes, false);
```
-## Drop Database
+## Manage tables
+
+### Create Table
-You can use the catalog to drop database.
+This is the same schema used by the Java reading and writing examples. Create
`my_db` first.
+For partitioned tables with fixed buckets, include partition columns in the
primary key.
+Tables that update a key across partitions require a different layout; see
+[cross-partition
upserts](../primary-key-table/data-distribution#cross-partitions-upsert).
```java
-import org.apache.paimon.catalog.Catalog;
+Identifier identifier = Identifier.create("my_db", "my_table");
+Schema schema = Schema.newBuilder()
+ .column("f0", DataTypes.STRING().notNull())
+ .column("f1", DataTypes.INT())
+ .primaryKey("f0")
+ .option("bucket", "2")
+ .build();
+catalog.createTable(identifier, schema, false);
+```
-public class DropDatabase {
+### Get Table
- public static void main(String[] args) {
- try {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- catalog.dropDatabase("my_db", false, true);
- } catch (Catalog.DatabaseNotEmptyException e) {
- // do something
- } catch (Catalog.DatabaseNotExistException e) {
- // do something
- }
- }
-}
+```java
+Table table = catalog.getTable(Identifier.create("my_db", "my_table"));
```
-## Alter Database
+Use the returned table with [Java Reads](java-reading) or [Java
Writes](java-writing).
-You can use the catalog to alter database's properties.(ps: only support hive
and jdbc catalog)
+### Determine Whether Table Exists
```java
-import java.util.ArrayList;
-import org.apache.paimon.catalog.Catalog;
-
-public class AlterDatabase {
-
- public static void main(String[] args) {
- try {
- Catalog catalog = CreateCatalog.createHiveCatalog();
- List<DatabaseChange> changes = new ArrayList<>();
- changes.add(DatabaseChange.setProperty("k1", "v1"));
- changes.add(DatabaseChange.removeProperty("k2"));
- catalog.alterDatabase("my_db", changes, true);
- } catch (Catalog.DatabaseNotExistException e) {
- // do something
- }
- }
-}
+boolean exists = catalog.tableExists(Identifier.create("my_db", "my_table"));
```
-## Determine Whether Table Exists
-
-You can use the catalog to determine whether the table exists
+### List Tables
```java
-import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.catalog.Identifier;
+List<String> tables = catalog.listTables("my_db");
+```
-public class TableExists {
+### Rename Table
- public static void main(String[] args) {
- Identifier identifier = Identifier.create("my_db", "my_table");
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- boolean exists = catalog.tableExists(identifier);
- }
-}
+```java
+catalog.renameTable(
+ Identifier.create("my_db", "my_table"),
+ Identifier.create("my_db", "renamed_table"),
+ false);
```
-## List Tables
+Subsequent operations must use the new identifier.
-You can use the catalog to list tables.
-
-```java
-import org.apache.paimon.catalog.Catalog;
+## Alter Table
-import java.util.List;
+Pass one `SchemaChange` or an ordered list of changes to `catalog.alterTable`.
The following example
+uses a separate table so that schema changes do not invalidate the Java
read/write walkthrough.
-public class ListTables {
+### Create a table for schema changes
- public static void main(String[] args) {
- try {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- List<String> tables = catalog.listTables("my_db");
- } catch (Catalog.DatabaseNotExistException e) {
- // do something
- }
- }
-}
+```java
+Identifier alterIdentifier = Identifier.create("my_db", "schema_example");
+Schema alterSchema = Schema.newBuilder()
+ .column("id", DataTypes.STRING().notNull())
+ .column("region", DataTypes.STRING().notNull())
+ .column("amount", DataTypes.INT().notNull())
+ .column("description", DataTypes.STRING())
+ .column("obsolete", DataTypes.STRING())
+ .column("details", DataTypes.ROW(
+ new DataField(0, "city", DataTypes.STRING().notNull())))
+ .primaryKey("id", "region")
+ .partitionKeys("region")
+ .option("bucket", "2")
+ .option("snapshot.num-retained.max", "20")
+ .build();
+catalog.createTable(alterIdentifier, alterSchema, false);
```
-## Drop Table
-
-You can use the catalog to drop table.
+### Apply schema and option changes
```java
-import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.catalog.Identifier;
+List<SchemaChange> changes = Arrays.asList(
+ SchemaChange.setOption("snapshot.time-retained", "2h"),
+ SchemaChange.removeOption("snapshot.num-retained.max"),
+ SchemaChange.addColumn("note", DataTypes.STRING(), "Optional note",
+ SchemaChange.Move.after("note", "description")),
+ SchemaChange.renameColumn("description", "description_text"),
+ SchemaChange.dropColumn("obsolete"),
+ SchemaChange.updateColumnComment(new String[] {"amount"}, "Order
amount"),
+ SchemaChange.updateColumnComment(new String[] {"details", "city"},
"City name"),
+ SchemaChange.updateColumnType("amount", DataTypes.BIGINT()),
+ SchemaChange.updateColumnPosition(SchemaChange.Move.first("amount")),
+ SchemaChange.updateColumnNullability(new String[] {"amount"}, true),
+ SchemaChange.updateColumnNullability(new String[] {"details", "city"},
true));
+catalog.alterTable(alterIdentifier, changes, false);
+
+// Load the updated schema before building new readers or writers.
+Table updatedTable = catalog.getTable(alterIdentifier);
+```
-public class DropTable {
+### Schema constraints
- public static void main(String[] args) {
- Identifier identifier = Identifier.create("my_db", "my_table");
- try {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- catalog.dropTable(identifier, false);
- } catch (Catalog.TableNotExistException e) {
- // do something
- }
- }
-}
-```
+- Newly added columns must be nullable.
+- Primary-key and partition-column types cannot be changed. Primary-key
nullability cannot be changed.
+- The example widens `INT` to `BIGINT` and relaxes `NOT NULL`. Tightening a
nullable column to
+ `NOT NULL` is disabled by default through
`alter-column-null-to-not-null.disabled`.
+- Nested field operations use a field path, for example `new String[]
{"details", "city"}`.
+ Supported type changes depend on the source and target types; replacing an
entire row type is
+ different from changing one nested field.
-## Rename Table
+See [schema evolution](../flink/sql-alter) for supported changes and their
constraints.
-You can use the catalog to rename a table.
+## Remove objects
-```java
-import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.catalog.Identifier;
+Drop operations remove catalog objects and can remove their stored data. Run
these separately from
+the read/write walkthrough, using the identifier that currently exists.
-public class RenameTable {
-
- public static void main(String[] args) {
- Identifier fromTableIdentifier = Identifier.create("my_db",
"my_table");
- Identifier toTableIdentifier = Identifier.create("my_db",
"test_table");
- try {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- catalog.renameTable(fromTableIdentifier, toTableIdentifier, false);
- } catch (Catalog.TableAlreadyExistException e) {
- // do something
- } catch (Catalog.TableNotExistException e) {
- // do something
- }
- }
-}
-```
+### Drop Table
-## Alter Table
+```java
+catalog.dropTable(Identifier.create("my_db", "renamed_table"), false);
+```
-You can use the catalog to alter a table, but you need to pay attention to the
following points.
+### Drop Database
-- Column %s cannot specify NOT NULL in the %s table.
-- Cannot update partition column type in the table.
-- Cannot change nullability of primary key.
-- If the type of the column is nested row type, update the column type is not
supported.
-- Update column to nested row type is not supported.
+Use `cascade=false` to reject dropping a nonempty database:
```java
-import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.schema.Schema;
-import org.apache.paimon.schema.SchemaChange;
-import org.apache.paimon.types.DataField;
-import org.apache.paimon.types.DataTypes;
-
-import com.google.common.collect.Lists;
+catalog.dropDatabase("my_db", false, false);
+```
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-
-public class AlterTable {
-
- public static void main(String[] args) {
- Identifier identifier = Identifier.create("my_db", "my_table");
-
- Map<String, String> options = new HashMap<>();
- options.put("bucket", "4");
-
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- catalog.createDatabase("my_db", false);
-
- try {
- catalog.createTable(
- identifier,
- new Schema(
- Lists.newArrayList(
- new DataField(0, "col1",
DataTypes.STRING(), "field1"),
- new DataField(1, "col2",
DataTypes.STRING(), "field2"),
- new DataField(2, "col3",
DataTypes.STRING(), "field3"),
- new DataField(3, "col4",
DataTypes.BIGINT(), "field4"),
- new DataField(
- 4,
- "col5",
- DataTypes.ROW(
- new DataField(
- 5, "f1",
DataTypes.STRING(), "f1"),
- new DataField(
- 6, "f2",
DataTypes.STRING(), "f2"),
- new DataField(
- 7, "f3",
DataTypes.STRING(), "f3")),
- "field5"),
- new DataField(8, "col6",
DataTypes.STRING(), "field6")),
- Lists.newArrayList("col1"), // partition keys
- Lists.newArrayList("col1", "col2"), // primary key
- options,
- "table comment"),
- false);
- } catch (Catalog.TableAlreadyExistException e) {
- // do something
- } catch (Catalog.DatabaseNotExistException e) {
- // do something
- }
+Use `cascade=true` only when you intend to remove the database and its tables:
- // add option
- SchemaChange addOption =
SchemaChange.setOption("snapshot.time-retained", "2h");
- // add column
- SchemaChange addColumn = SchemaChange.addColumn("col1_after",
DataTypes.STRING());
- // add a column after col1
- SchemaChange.Move after = SchemaChange.Move.after("col1_after",
"col1");
- SchemaChange addColumnAfterField =
- SchemaChange.addColumn("col7", DataTypes.STRING(), "", after);
- // rename column
- SchemaChange renameColumn = SchemaChange.renameColumn("col3",
"col3_new_name");
- // drop column
- SchemaChange dropColumn = SchemaChange.dropColumn("col6");
- // update column comment
- SchemaChange updateColumnComment =
- SchemaChange.updateColumnComment(new String[]{"col4"}, "col4
field");
- // update nested column comment
- SchemaChange updateNestedColumnComment =
- SchemaChange.updateColumnComment(new String[]{"col5", "f1"},
"col5 f1 field");
- // update column type
- SchemaChange updateColumnType = SchemaChange.updateColumnType("col4",
DataTypes.DOUBLE());
- // update column position, you need to pass in a parameter of type Move
- SchemaChange updateColumnPosition =
-
SchemaChange.updateColumnPosition(SchemaChange.Move.first("col4"));
- // update column nullability
- SchemaChange updateColumnNullability =
- SchemaChange.updateColumnNullability(new String[]{"col4"},
false);
- // update nested column nullability
- SchemaChange updateNestedColumnNullability =
- SchemaChange.updateColumnNullability(new String[]{"col5",
"f2"}, false);
-
- SchemaChange[] schemaChanges =
- new SchemaChange[]{
- addOption,
- removeOption,
- addColumn,
- addColumnAfterField,
- renameColumn,
- dropColumn,
- updateColumnComment,
- updateNestedColumnComment,
- updateColumnType,
- updateColumnPosition,
- updateColumnNullability,
- updateNestedColumnNullability
- };
- try {
- catalog.alterTable(identifier, Arrays.asList(schemaChanges),
false);
- } catch (Catalog.TableNotExistException e) {
- // do something
- } catch (Catalog.ColumnAlreadyExistException e) {
- // do something
- } catch (Catalog.ColumnNotExistException e) {
- // do something
- }
- }
-}
+```java
+catalog.dropDatabase("my_db", false, true);
```
diff --git a/docs/docs/program-api/cpp-api.md b/docs/docs/program-api/cpp-api.md
index f9962ccf9e..5d208422e1 100644
--- a/docs/docs/program-api/cpp-api.md
+++ b/docs/docs/program-api/cpp-api.md
@@ -1,9 +1,8 @@
---
-title: "Cpp API"
-sidebar_position: 6
+title: "C++ API"
+sidebar_position: 8
---
-
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
@@ -23,15 +22,24 @@ specific language governing permissions and limitations
under the License.
-->
-# Cpp API
+# C++ API
+
+[Paimon C++](https://github.com/apache/paimon-cpp) provides native table
access for C++ applications
+and engines. It exchanges columnar data through the Arrow C Data Interface.
+
+This walkthrough follows the same workflow as the Java API: create a catalog
and table, prepare and
+commit a batch, then plan splits and read them. The snippets are function-body
fragments returning
+`paimon::Status` (place includes at file scope); `PAIMON_RETURN_NOT_OK` and
`PAIMON_ASSIGN_OR_RAISE` propagate failures to its caller.
+The `PrepareData` helper returns `arrow::Result` and uses Arrow's error macros
instead.
-[Paimon C++](https://github.com/apache/paimon-cpp.git) is a high-performance
C++ implementation of Apache Paimon.
-Paimon C++ aims to provide a native, high-performance and extensible
implementation
-that allows native engines to access the Paimon datalake format with maximum
efficiency.
+For complete headers and a runnable application, start with the
+[C++ examples](https://paimon.apache.org/docs/cpp/examples/index.html). The
C++ project has its own
+release cycle; use its build instructions and API reference for the version
you select.
## Environment Settings
-You can checkout the [document](https://paimon.apache.org/docs/cpp/index.html)
for more details about environment settings.
+Follow the [C++ build guide](https://paimon.apache.org/docs/cpp/building.html)
to install prerequisites
+and select optional filesystem, file-format, and catalog components. A basic
source build is:
```sh
git clone https://github.com/apache/paimon-cpp.git
@@ -45,36 +53,37 @@ make install
## Create Catalog
-Before coming into contact with the Table, you need to create a Catalog.
+Create a filesystem catalog for a warehouse. Reuse these options and
identifiers in the following
+fragments; choose a fresh table name when running the walkthrough again.
```c++
#include "paimon/catalog/catalog.h"
-// Note that keys and values are all string
+const std::string root_path = "/tmp/paimon-cpp-warehouse";
+const std::string db_name = "my_db";
+const std::string table_name = "my_table";
std::map<std::string, std::string> options;
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::Catalog> catalog,
paimon::Catalog::Create(root_path, options));
```
-Current C++ Paimon only supports filesystem catalog. In the future, we will
support REST catalog.
-See [Catalog](../concepts/catalog).
-
-You can use the catalog to create table for writing data.
+C++ also supports a REST catalog when built with `PAIMON_ENABLE_REST=ON`.
+See the [C++ catalog
guide](https://paimon.apache.org/docs/cpp/user_guide/catalog.html) for options
+and supported operations. For REST-managed tables, use the table location
returned by the catalog;
+the path construction below is specific to the filesystem catalog.
## Create Database
-Table is located in a database. If you want to create table in a new database,
you should create it.
+Create the database before the table:
```c++
-PAIMON_RETURN_NOT_OK(catalog->CreateDatabase('database_name', options,
/*ignore_if_exists=*/false));
+PAIMON_RETURN_NOT_OK(catalog->CreateDatabase(db_name, options,
/*ignore_if_exists=*/false));
```
## Create Table
-Table schema contains fields definition, partition keys, primary keys, table
options.
-The field definition is described by `Arrow::Schema`. All arguments except
fields definition are optional.
-
-for example:
+Define fields using an `arrow::Schema`, then export it through the Arrow C
Data Interface.
+This example creates an unpartitioned append table without primary keys.
```c++
arrow::FieldVector fields = {
@@ -101,11 +110,13 @@ See [Data
Types](https://paimon.apache.org/docs/cpp/user_guide/data_types.html)
## Batch Write
-Paimon table write is Two-Phase Commit, you can write many times, but once
committed, no more data can be written.
-C++ Paimon uses Apache Arrow as [in-memory format], check out
[document](https://paimon.apache.org/docs/cpp/user_guide/arrow.html)
-for more details.
+First construct an Arrow batch, then write it and prepare commit messages. The
committer publishes
+the prepared changes. In a distributed application, collect messages from the
participating writers
+before committing. See the [memory format
guide](https://paimon.apache.org/docs/cpp/user_guide/arrow.html)
+for ownership and Arrow conversion details.
+
+### Build a batch
-for example:
```c++
arrow::Result<std::shared_ptr<arrow::StructArray>> PrepareData(const
arrow::FieldVector& fields) {
arrow::StringBuilder f0_builder;
@@ -135,6 +146,8 @@ arrow::Result<std::shared_ptr<arrow::StructArray>>
PrepareData(const arrow::Fiel
}
```
+### Write and commit
+
```c++
std::string table_path = root_path + "/" + db_name + ".db/" + table_name;
std::string commit_user = "some_commit_user";
@@ -172,39 +185,39 @@ PAIMON_RETURN_NOT_OK(committer->Commit(commit_message));
## Batch Read
-### Predicate pushdown
-
-A `ReadContextBuilder` is used to pass context to reader, push down and filter
is done by reader.
+Configure the reader, plan the splits, then consume each batch. When
distributing reads, plan once
+and assign splits to reader tasks.
-```c++
-ReadContextBuilder read_context_builder(table_path);
-```
+### Predicate pushdown
-You can use `PredicateBuilder` to build filters and pushdown them by
`ReadContextBuilder`:
+Use `ReadContextBuilder` to configure the reader.
`EnablePredicateFilter(true)` requests row-level
+filtering as well as any pruning supported by the reader:
```c++
-# Example filter: 'f3' > 12.0 OR 'f1' == 1
+// Example filter: 'f3' > 12.0 OR 'f1' == 1
PAIMON_ASSIGN_OR_RAISE(
auto predicate,
- PredicateBuilder::Or(
- {PredicateBuilder::GreaterThan(/*field_index=*/3, /*field_name=*/"f3",
- FieldType::DOUBLE,
Literal(static_cast<double>(12.0))),
- PredicateBuilder::Equal(/*field_index=*/1, /*field_name=*/"f1",
FieldType::INT,
- Literal(1))}));
-ReadContextBuilder read_context_builder(table_path);
+ paimon::PredicateBuilder::Or({
+ paimon::PredicateBuilder::GreaterThan(
+ /*field_index=*/3, /*field_name=*/"f3",
+ paimon::FieldType::DOUBLE, paimon::Literal(12.0)),
+ paimon::PredicateBuilder::Equal(
+ /*field_index=*/1, /*field_name=*/"f1",
+ paimon::FieldType::INT, paimon::Literal(1))}));
+paimon::ReadContextBuilder read_context_builder(table_path);
read_context_builder.SetPredicate(predicate).EnablePredicateFilter(true);
```
-You can also pushdown projection by `ReadContextBuilder`:
+Set the projected fields on the same read context:
```c++
-# select f3 and f2 columns
-read_context_builder.SetReadSchema({"f3", "f1", "f2"});
+// Return f3, f1, and f2, in that order
+read_context_builder.SetReadFieldNames({"f3", "f1", "f2"});
```
### Generate Splits
-Then you can step into Scan Plan stage to get `splits`:
+Create a scan plan to discover the splits to read:
```c++
// scan
@@ -217,11 +230,12 @@ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::Plan>
plan, scanner->CreatePlan()
auto splits = plan->Splits();
```
-Finally, you can read data from the `splits` to arrow format.
+Pass the planned splits to a table reader to obtain Arrow batches.
### Read Apache Arrow
-This requires `C++ Arrow` to be installed.
+Import each returned batch into Arrow C++ objects. This example collects the
batches in memory;
+for large results, process batches as they arrive instead.
```c++
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<paimon::ReadContext> read_context,
diff --git a/docs/docs/program-api/file-cache.mdx
b/docs/docs/program-api/file-cache.mdx
index eff163ada8..8c58a398bc 100644
--- a/docs/docs/program-api/file-cache.mdx
+++ b/docs/docs/program-api/file-cache.mdx
@@ -1,6 +1,6 @@
---
title: "Local Cache"
-sidebar_position: 8
+sidebar_position: 10
---
import Tabs from '@theme/Tabs';
@@ -27,15 +27,22 @@ under the License.
# Local Cache
-When reading files from remote storage (S3, OSS, HDFS, etc.), each seek+read
goes over the network. Paimon provides a block-level local cache that
transparently caches file reads, significantly reducing remote I/O for repeated
access patterns.
+Paimon's local cache stores file blocks on disk or in memory to reduce
repeated reads from
+storage such as S3, OSS, and HDFS. Configure it on the catalog that loads your
tables.
-The cache supports two modes:
-- **Disk cache**: when `local-cache.dir` is configured, blocks are cached on
local disk.
-- **Memory cache**: when `local-cache.dir` is not configured, blocks are
cached in memory.
+| Mode | When it is selected | Where blocks live |
+| --- | --- | --- |
+| Memory | Cache enabled without `local-cache.dir` | Process memory |
+| Disk | Cache enabled with `local-cache.dir` | The configured local directory
|
+
+Set an explicit size limit for your workload. The cache is disabled by
default, and enabling it
+does not automatically cache data files: the default whitelist covers metadata
and global indexes.
+
+
## Cached File Types
-The cache classifies files by type. By default, only `meta` and `global-index`
types are cached. You can customize this via the `local-cache.whitelist` option.
+The cache classifies files by path. Set `local-cache.whitelist` to select the
file types to cache.
| File Type | Config Name | Examples | Default Cached |
|-----------|-------------|----------|----------------|
@@ -45,7 +52,8 @@ The cache classifies files by type. By default, only `meta`
and `global-index` t
| DATA | data | Data files (ORC, Parquet, etc.) | No |
| FILE_INDEX | file-index | Data-file level bloom filter, bitmap | No |
-All file types can be added to the whitelist. The default whitelist is
`meta,global-index`.
+The default whitelist is `meta,global-index`. The mutable `LATEST` and
`EARLIEST` hint files bypass
+the cache even though they are classified as metadata.
## Enable Cache
@@ -56,24 +64,27 @@ This is a catalog-level option. Configure it when creating
the catalog:
<TabItem value="java" label="Java">
```java
+import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
Options options = new Options();
options.set("warehouse", "s3://my-bucket/warehouse");
options.set("local-cache.enabled", "true");
// optional: use disk cache by specifying a directory
options.set("local-cache.dir", "/tmp/paimon-cache");
-// optional: customize limits
+// Set a limit for cached blocks
options.set("local-cache.max-size", "2gb");
options.set("local-cache.block-size", "1mb");
CatalogContext context = CatalogContext.create(options);
-Catalog catalog = CatalogFactory.createCatalog(context);
-
-// All tables from this catalog will use the cache
-Table table = catalog.getTable(Identifier.create("my_db", "my_table"));
+try (Catalog catalog = CatalogFactory.createCatalog(context)) {
+ Table table = catalog.getTable(Identifier.create("my_db", "my_table"));
+ // Read table data here; eligible file reads use the cache.
+}
```
</TabItem>
@@ -88,14 +99,14 @@ options = {
"local-cache.enabled": "true",
# optional: use disk cache by specifying a directory
"local-cache.dir": "/tmp/paimon-cache",
- # optional: customize limits
+ # Set a limit for cached blocks
"local-cache.max-size": "2gb",
"local-cache.block-size": "1mb",
}
catalog = pypaimon.create_catalog(options)
-# All tables from this catalog will use the cache
+# Eligible file reads from tables loaded by this catalog use the cache
table = catalog.get_table("db.my_table")
```
@@ -103,28 +114,57 @@ table = catalog.get_table("db.my_table")
</Tabs>
+The snippets assume the catalog and table already exist and storage
credentials are configured.
+The Java snippet is a method-body fragment; declare or handle the catalog
exceptions as in the
+[Java setup](java-api).
+
## Cache Options
| Option | Type | Default | Description |
|--------|------|---------|-------------|
| `local-cache.enabled` | Boolean | false | Whether to enable local block
cache for file reads. |
| `local-cache.dir` | String | (none) | Directory for storing cached blocks on
disk. If not configured, memory cache is used. |
-| `local-cache.max-size` | MemorySize | unlimited | Maximum total size of the
cache. When exceeded, the least recently used blocks are evicted. |
+| `local-cache.max-size` | MemorySize | See below | Maximum cached block bytes
per cache manager. Least recently used blocks are evicted when the limit is
exceeded. |
| `local-cache.block-size` | MemorySize | 1 mb | Block size for caching. Files
are logically divided into fixed-size blocks and cached independently. |
| `local-cache.whitelist` | String | meta,global-index | Comma-separated list
of file types to cache. Supported values: `meta`, `global-index`,
`bucket-index`, `data`, `file-index`. |
+When `local-cache.max-size` is omitted, Java has no configured size limit.
PyPaimon uses a 256 MiB
+memory limit or a 10 GiB disk limit. Set the option explicitly when you want
the same limit across
+clients. The limit accounts for cached block bytes, not all reader buffers or
process memory.
+
## How It Works
-- Files are logically divided into fixed-size blocks (default 1 MB).
-- On the first read, blocks are downloaded from remote storage and cached
locally (on disk or in memory).
-- Subsequent reads of the same block are served from the local cache, skipping
remote I/O.
-- When using disk cache, cache files are keyed by remote file path and block
offset, so they persist across process restarts and can be reused.
-- When the cache exceeds `max-size`, the least recently used blocks are
evicted automatically.
+1. The reader requests bytes from an eligible file. The cache maps the byte
range to fixed-size
+ blocks, using 1 MiB blocks by default.
+2. A hit returns the cached block. A miss reads the block from storage and
adds it to the cache.
+3. Least recently used blocks are evicted when a configured limit is exceeded.
+
+A cache hit avoids fetching that block's contents again; it does not eliminate
all remote metadata
+operations. In Java disk mode, opening a file still obtains its status. The
disk cache identifies a
+file by path, length, and modification time, and separates entries by block
size. Existing entries
+can be reused after restart when this identity matches. Memory caches do not
survive process restarts.
+
+Caching applies to reads. It does not change commit semantics or replace
durable table storage.
## Cache Lifecycle
-The cache is created and managed by the Catalog. All tables obtained from the
same catalog share a single cache instance. The cache lives as long as the
Catalog object is reachable — no explicit close is needed.
+### Java applications and distributed workers
+
+A catalog creates the cache used by the tables it loads. After a
catalog-created `FileIO` is
+serialized and deserialized on a worker, it retains its cache configuration
and lazily creates or
+reuses a cache manager in that worker JVM.
+
+Deserialized wrappers with matching directory, maximum size, and block size
share a JVM-local cache
+manager and its size limit. Memory entries remain isolated by the originating
wrapper's namespace.
+The limit is local to that manager, not a cluster-wide budget. In disk mode
the directory must be
+available and writable on each worker that uses it.
+
+Close readers and other owned resources when finished. Closing a deserialized
`FileIO` releases its
+reference to the shared manager; the last reference releases the shared entry.
A manually constructed
+`CachingFileIO` without serialized cache configuration cannot recreate a cache
after deserialization.
-In distributed computing frameworks (Flink, Spark), the `FileIO` is serialized
and shipped to task managers. After deserialization, the cache is **not**
recreated — reads fall through directly to the remote storage. This is by
design: the cache lifecycle is bound to the Catalog that created it, and a
deserialized `FileIO` is no longer managed by any Catalog.
+### PyPaimon workers
-If you need caching on task managers, create a new Catalog with cache options
enabled on each worker node.
+PyPaimon's cache is removed when `CachingFileIO` is pickled. An unpickled copy
reads directly from
+storage. Create a catalog with cache options on each Python worker when you
need caching there.
+The Java worker lifecycle above does not apply to Python serialization.
diff --git a/docs/docs/program-api/flink-api.mdx
b/docs/docs/program-api/flink-api.mdx
index 2ac1775a85..20010aa507 100644
--- a/docs/docs/program-api/flink-api.mdx
+++ b/docs/docs/program-api/flink-api.mdx
@@ -1,6 +1,6 @@
---
title: "Flink API"
-sidebar_position: 2
+sidebar_position: 6
---
import Stable from '@site/src/components/Stable';
@@ -27,11 +27,19 @@ under the License.
# Flink API
-:::info
+Use the Flink builders to connect a `DataStream` to a Paimon table. The sink
integrates writer
+routing, checkpoints, and commits with Flink. Use `RichCdcSinkBuilder` when
incoming records also
+carry schema changes.
-If possible, recommend using Flink SQL or Spark SQL, or simply use SQL APIs in
programs.
+| Input or task | API |
+| --- | --- |
+| Flink `Row` values with a known schema | `FlinkSinkBuilder.forRow` |
+| Flink internal `RowData` values | `FlinkSinkBuilder.forRowData` |
+| Read a table as a `DataStream<Row>` | `FlinkSourceBuilder.buildForRow` |
+| Ingest typed CDC records with schema evolution | `RichCdcSinkBuilder` |
-:::
+For SQL transformations, you can also convert between DataStream and the Table
API.
+See [DataStream API
Integration](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/data_stream_api/).
## Dependency
@@ -64,14 +72,21 @@ Or download the jar file:
</Unstable>
-Please choose your Flink version.
+Match the Paimon artifact and Flink dependencies to the Flink version you run.
See
+[Flink installation](../flink/installation) for connector and Hadoop runtime
setup.
-Paimon relies on Hadoop environment, you should add hadoop classpath or
bundled jar.
+## Prepare a table
-Not only DataStream API, you can also read or write to Paimon tables by the
conversion between DataStream and Table in Flink.
-See [DataStream API
Integration](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/data_stream_api/).
+The read and write examples use `my_db.my_table` from the [Java API
setup](java-api#create-table),
+with columns `f0 STRING` and `f1 INT`, primary key `f0`, and two fixed
buckets. Use the same warehouse
+in each example. In a cluster, choose a
[filesystem](../maintenance/filesystems) accessible to all
+workers instead of the local path shown here.
+
+## Write to Table
-## Write to Table
+Declare the input's field names and types in the same order as the Paimon
schema. The example
+updates Alice's value from 12 to 100. Enable checkpointing for an unbounded
streaming source so
+that the sink can publish data as checkpoints complete.
```java
import org.apache.paimon.catalog.Catalog;
@@ -92,10 +107,9 @@ import org.apache.flink.types.RowKind;
public class WriteToTable {
public static void writeTo() throws Exception {
- // create environments of both APIs
+ // Create the Flink execution environment.
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
- // for CONTINUOUS_UNBOUNDED source, set checkpoint interval
- // env.enableCheckpointing(60_000);
+ env.enableCheckpointing(60_000);
// create a changelog DataStream
DataStream<Row> input =
@@ -106,34 +120,35 @@ public class WriteToTable {
Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100))
.returns(
Types.ROW_NAMED(
- new String[] {"name", "age"},
Types.STRING, Types.INT));
+ new String[] {"f0", "f1"},
Types.STRING, Types.INT));
// get table from catalog
Options catalogOptions = new Options();
- catalogOptions.set("warehouse", "/path/to/warehouse");
- Catalog catalog =
FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
- Table table = catalog.getTable(Identifier.create("my_db", "T"));
-
- DataType inputType =
- DataTypes.ROW(
- DataTypes.FIELD("name", DataTypes.STRING()),
- DataTypes.FIELD("age", DataTypes.INT()));
- FlinkSinkBuilder builder = new FlinkSinkBuilder(table).forRow(input,
inputType);
-
- // set sink parallelism
- // builder.parallelism(_your_parallelism)
-
- // set overwrite mode
- // builder.overwrite(...)
-
- builder.build();
- env.execute();
+ catalogOptions.set("warehouse", "file:///tmp/paimon-api-warehouse");
+ try (Catalog catalog =
FlinkCatalogFactory.createPaimonCatalog(catalogOptions)) {
+ Table table = catalog.getTable(Identifier.create("my_db",
"my_table"));
+
+ DataType inputType =
+ DataTypes.ROW(
+ DataTypes.FIELD("f0", DataTypes.STRING()),
+ DataTypes.FIELD("f1", DataTypes.INT()));
+ FlinkSinkBuilder builder = new
FlinkSinkBuilder(table).forRow(input, inputType);
+
+ builder.build();
+ env.execute();
+ }
}
}
```
+Use `builder.parallelism(...)` to set sink parallelism. Select
`builder.overwrite()` only for an
+intentional overwrite; see [write semantics](../flink/sql-write).
+
## Read from Table
+This example performs a bounded read of the current table state. Use
`sourceBounded(false)` for a
+continuous source and choose the appropriate [streaming read
mode](../flink/sql-query).
+
```java
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
@@ -145,55 +160,66 @@ import org.apache.paimon.table.Table;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
public class ReadFromTable {
public static void readFrom() throws Exception {
- // create environments of both APIs
+ // Create the Flink execution environment.
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// get table from catalog
Options catalogOptions = new Options();
- catalogOptions.set("warehouse", "/path/to/warehouse");
- Catalog catalog =
FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
- Table table = catalog.getTable(Identifier.create("my_db", "T"));
-
- // table =
table.copy(Collections.singletonMap("scan.file-creation-time-millis", "..."));
-
- FlinkSourceBuilder builder = new FlinkSourceBuilder(table).env(env);
-
- // builder.sourceBounded(true);
- // builder.projection(...);
- // builder.predicate(...);
- // builder.limit(...);
- // builder.sourceParallelism(...);
-
- DataStream<Row> dataStream = builder.buildForRow();
-
- // use this datastream
- dataStream.executeAndCollect().forEachRemaining(System.out::println);
-
- // prints:
- // +I[Bob, 12]
- // +I[Alice, 12]
- // -U[Alice, 12]
- // +U[Alice, 14]
+ catalogOptions.set("warehouse", "file:///tmp/paimon-api-warehouse");
+ try (Catalog catalog =
FlinkCatalogFactory.createPaimonCatalog(catalogOptions)) {
+ Table table = catalog.getTable(Identifier.create("my_db",
"my_table"));
+
+ FlinkSourceBuilder builder = new FlinkSourceBuilder(table)
+ .env(env)
+ .sourceBounded(true);
+
+ DataStream<Row> dataStream = builder.buildForRow();
+
+ try (CloseableIterator<Row> rows = dataStream.executeAndCollect())
{
+ rows.forEachRemaining(System.out::println);
+ }
+ }
}
}
```
-## Cdc ingestion Table
+After running the write example against the fresh sample table, the bounded
result contains
+Alice with value 100 and Bob with value 5. Output order is not guaranteed. A
continuous read's
+row kinds depend on the table's [changelog
configuration](../primary-key-table/changelog-producer).
+
+Use `projection`, `predicate`, `limit`, and `sourceParallelism` on the source
builder to customize
+the scan. Predicate field indexes refer to the table schema.
+
+## CDC ingestion with schema evolution {#cdc-ingestion-table}
+
+`RichCdcRecord` carries field names, types, and string-encoded values.
`RichCdcSinkBuilder` uses the
+catalog loader to apply schema changes and write the records. This also
supports adding columns
+to a [partial-update table](../primary-key-table/merge-engine/partial-update).
+
+Create this separate table first, using a Flink SQL catalog pointing to the
same warehouse:
-Paimon supports ingest data into Paimon tables with schema evolution.
-- You can use Java API to write cdc records into Paimon Tables.
-- You can write records to Paimon's partial-update table with adding columns
dynamically.
+```sql
+CREATE DATABASE IF NOT EXISTS my_db;
+CREATE TABLE my_db.cdc_orders (
+ order_id BIGINT,
+ price DOUBLE,
+ PRIMARY KEY (order_id) NOT ENFORCED
+) WITH ('bucket' = '2');
+```
-Here is an example to use `RichCdcSinkBuilder` API:
+The second record below introduces `dt`. Supply a serializable `CatalogLoader`
so runtime operators
+can access the same catalog and evolve the schema.
```java
+import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogLoader;
-import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.sink.cdc.RichCdcRecord;
import org.apache.paimon.flink.sink.cdc.RichCdcSinkBuilder;
import org.apache.paimon.options.Options;
@@ -209,8 +235,7 @@ public class WriteCdcToTable {
public static void writeTo() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
- // for CONTINUOUS_UNBOUNDED source, set checkpoint interval
- // env.enableCheckpointing(60_000);
+ env.enableCheckpointing(60_000);
DataStream<RichCdcRecord> dataStream =
env.fromElements(
@@ -225,20 +250,22 @@ public class WriteCdcToTable {
.field("dt", DataTypes.TIMESTAMP(),
"2023-06-12 20:21:12")
.build());
- Identifier identifier = Identifier.create("my_db", "T");
+ Identifier identifier = Identifier.create("my_db", "cdc_orders");
Options catalogOptions = new Options();
- catalogOptions.set("warehouse", "/path/to/warehouse");
+ catalogOptions.set("warehouse", "file:///tmp/paimon-api-warehouse");
CatalogLoader catalogLoader =
() -> FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
- Table table = catalogLoader.load().getTable(identifier);
+ try (Catalog catalog = catalogLoader.load()) {
+ Table table = catalog.getTable(identifier);
- new RichCdcSinkBuilder(table)
- .forRichCdcRecord(dataStream)
- .identifier(identifier)
- .catalogLoader(catalogLoader)
- .build();
+ new RichCdcSinkBuilder(table)
+ .forRichCdcRecord(dataStream)
+ .identifier(identifier)
+ .catalogLoader(catalogLoader)
+ .build();
- env.execute();
+ env.execute();
+ }
}
}
-```
\ No newline at end of file
+```
diff --git a/docs/docs/program-api/index.md b/docs/docs/program-api/index.md
index 59c775d0b4..890a42ea1e 100644
--- a/docs/docs/program-api/index.md
+++ b/docs/docs/program-api/index.md
@@ -21,3 +21,46 @@ KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
+
+# Program API
+
+Use Paimon's program APIs to manage catalogs, embed table reads and writes in
an application,
+or build a connector for a processing engine. Start with the interface that
matches your application.
+
+## Choose an API
+
+| What you want to do | Start here |
+| --- | --- |
+| Read and write tables in a standalone Java application | [Java
API](java-api) |
+| Create databases and tables, or change schemas | [Catalog API](catalog-api) |
+| Build a Flink DataStream job or ingest records with schema evolution |
[Flink API](flink-api) |
+| Call a REST catalog from a lightweight Java client | [REST Java
Client](rest-api) |
+| Integrate a native C++ engine | [C++ API](cpp-api) |
+| Access Paimon from Rust | [Rust API](rust-api) |
+| Work with Python, Arrow, or AI datasets | [PyPaimon](../pypaimon/) |
+| Reduce repeated file reads | [Local Cache](file-cache) |
+
+For SQL applications, start with [Flink](../flink/quick-start) or
[Spark](../spark/quick-start).
+These integrations handle execution, data distribution, and recovery for you.
+
+
+
+## Follow the Java workflow
+
+1. **Set up the client.** Add the [dependency and create a
catalog](java-api#dependency).
+2. **Create or load a table.** Define its schema through the [Catalog
API](catalog-api).
+3. **Read or write data.** Follow [Java Reads](java-reading) or [Java
Writes](java-writing)
+ for batch and streaming examples.
+4. **Integrate with your runtime.** Distribute splits and writer input, close
resources, and
+ coordinate checkpoints and commits. Use [types and predicates](java-types)
to convert records
+ and construct filters.
+
+## Understand the boundaries
+
+A **catalog** resolves table names and manages metadata. A **table** supplies
builders for reads
+and writes. A **scan** plans splits; readers consume those splits. Writers
prepare file changes;
+a committer publishes them in snapshots.
+
+The low-level Java API exposes these building blocks. A custom distributed
application must provide
+scheduling, writer routing, and recovery. The [Flink builders](flink-api)
connect Paimon to Flink's runtime.
+The [REST Java client](rest-api) handles catalog requests; use a table API to
read or write rows.
diff --git a/docs/docs/program-api/java-api.mdx
b/docs/docs/program-api/java-api.mdx
index 0f58070037..b03de1c982 100644
--- a/docs/docs/program-api/java-api.mdx
+++ b/docs/docs/program-api/java-api.mdx
@@ -1,6 +1,6 @@
---
title: "Java API"
-sidebar_position: 3
+sidebar_position: 1
---
import Stable from '@site/src/components/Stable';
@@ -27,11 +27,17 @@ under the License.
# Java API
-:::info
+Use the Java API to access Paimon tables without a processing engine, or to
implement an engine
+integration. This page sets up a catalog and a sample table used by the
reading and writing guides.
-If possible, recommend using computing engines such as Flink SQL or Spark SQL.
+| Task | Guide |
+| --- | --- |
+| Manage databases, tables, and schemas | [Catalog API](catalog-api) |
+| Plan splits, filter rows, and follow new snapshots | [Java
Reads](java-reading) |
+| Prepare and commit batch or streaming writes | [Java Writes](java-writing) |
+| Convert row values and build predicates | [Types and Predicates](java-types)
|
-:::
+For a Flink application, use the [Flink API](flink-api) so the engine can
manage checkpoints and commits.
## Dependency
@@ -57,11 +63,14 @@ Or download the jar file:
</Unstable>
-Paimon relies on Hadoop environment, you should add hadoop classpath or
bundled jar.
+Add the Hadoop libraries required by your environment to the runtime
classpath. For remote storage,
+configure the corresponding [filesystem dependencies and
credentials](../maintenance/filesystems).
## Create Catalog
-Before coming into contact with the Table, you need to create a Catalog.
+`CatalogFactory` selects a catalog from the supplied options. The helper below
uses a local
+filesystem warehouse. Save it as `CreateCatalog.java` alongside the example
classes, and change the
+warehouse path for your environment.
```java
import org.apache.paimon.catalog.Catalog;
@@ -73,29 +82,31 @@ import org.apache.paimon.options.Options;
public class CreateCatalog {
public static Catalog createFilesystemCatalog() {
- CatalogContext context = CatalogContext.create(new Path("..."));
+ CatalogContext context =
+ CatalogContext.create(new
Path("file:///tmp/paimon-api-warehouse"));
return CatalogFactory.createCatalog(context);
}
public static Catalog createHiveCatalog() {
- // Paimon Hive catalog relies on Hive jars
- // You should add hive classpath or hive bundled jar.
+ // Add the Hive client libraries and configure these paths for your
environment.
Options options = new Options();
- options.set("warehouse", "...");
+ options.set("warehouse", "hdfs:///path/to/warehouse");
options.set("metastore", "hive");
- options.set("uri", "...");
- options.set("hive-conf-dir", "...");
- options.set("hadoop-conf-dir", "...");
- CatalogContext context = CatalogContext.create(options);
- return CatalogFactory.createCatalog(context);
+ options.set("uri", "thrift://localhost:9083");
+ options.set("hive-conf-dir", "/path/to/hive/conf");
+ options.set("hadoop-conf-dir", "/path/to/hadoop/conf");
+ return CatalogFactory.createCatalog(CatalogContext.create(options));
}
}
```
+Keep the catalog open while using its tables, and close it when the
application finishes.
+See [Catalogs](../concepts/catalog) for other catalog implementations and
their configuration.
+
## Create Table
-You can use the catalog to create tables. The created tables are persistence
in the file system.
-Next time you can directly obtain these tables.
+Run this example once in a fresh warehouse. It creates `my_db.my_table` with a
string primary key
+`f0`, an integer value `f1`, and two fixed buckets. The following guides use
this same schema.
```java
import org.apache.paimon.catalog.Catalog;
@@ -105,376 +116,75 @@ import org.apache.paimon.types.DataTypes;
public class CreateTable {
- public static void main(String[] args) {
- Schema.Builder schemaBuilder = Schema.newBuilder();
- schemaBuilder.primaryKey("f0", "f1");
- schemaBuilder.partitionKeys("f1");
- schemaBuilder.column("f0", DataTypes.STRING());
- schemaBuilder.column("f1", DataTypes.INT());
- Schema schema = schemaBuilder.build();
-
- Identifier identifier = Identifier.create("my_db", "my_table");
- try {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- catalog.createTable(identifier, schema, false);
- } catch (Catalog.TableAlreadyExistException e) {
- // do something
- } catch (Catalog.DatabaseNotExistException e) {
- // do something
+ public static void main(String[] args) throws Exception {
+ Schema schema = Schema.newBuilder()
+ .column("f0", DataTypes.STRING().notNull())
+ .column("f1", DataTypes.INT())
+ .primaryKey("f0")
+ .option("bucket", "2")
+ .build();
+
+ try (Catalog catalog = CreateCatalog.createFilesystemCatalog()) {
+ catalog.createDatabase("my_db", true);
+ catalog.createTable(Identifier.create("my_db", "my_table"),
schema, false);
}
}
}
```
+`true` in `createDatabase` ignores an existing database; `false` in
`createTable` reports an
+existing table. This avoids silently reusing a table with a different schema.
+For an append table, omit the primary key and choose the appropriate
+[append table layout](../append-table/).
+
## Get Table
-The `Table` interface provides access to the table metadata and tools to read
and write table.
+Load the table by identifier within the catalog's lifetime:
```java
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.table.Table;
-public class GetTable {
-
- public static Table getTable() {
- Identifier identifier = Identifier.create("my_db", "my_table");
- try {
- Catalog catalog = CreateCatalog.createFilesystemCatalog();
- return catalog.getTable(identifier);
- } catch (Catalog.TableNotExistException e) {
- // do something
- throw new RuntimeException("table not exist");
- }
- }
-}
-```
-
-## Batch Read
-
-For relatively small amounts of data, or for data that has undergone
projection and filtering,
-you can directly use a standalone program to read the table data.
-
-But if the data volume of the table is relatively large, you can distribute
splits to different tasks for reading.
-
-The reading is divided into two stages:
-
-1. Scan Plan: Generate plan splits in a global node ('Coordinator', or named
'Driver').
-2. Read Split: Read split in distributed tasks.
-
-```java
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateBuilder;
-import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.Split;
-import org.apache.paimon.table.source.TableRead;
-import org.apache.paimon.types.DataTypes;
-import org.apache.paimon.types.RowType;
-
-import com.google.common.collect.Lists;
-
-import java.util.List;
-
-public class ReadTable {
-
- public static void main(String[] args) throws Exception {
- // 1. Create a ReadBuilder and push filter (`withFilter`)
- // and projection (`withProjection`) if necessary
- Table table = GetTable.getTable();
-
- PredicateBuilder builder =
- new PredicateBuilder(RowType.of(DataTypes.STRING(),
DataTypes.INT()));
- Predicate notNull = builder.isNotNull(0);
- Predicate greaterOrEqual = builder.greaterOrEqual(1, 12);
-
- int[] projection = new int[]{0, 1};
-
- ReadBuilder readBuilder =
- table.newReadBuilder()
- .withProjection(projection)
- .withFilter(Lists.newArrayList(notNull,
greaterOrEqual));
-
- // 2. Plan splits in 'Coordinator' (or named 'Driver')
- List<Split> splits = readBuilder.newScan().plan().splits();
-
- // 3. Distribute these splits to different tasks
-
- // 4. Read a split in task
- // You can use executeFilter to do filter per record.
- // By default, only capable of performing coarse-grained filtering.
- TableRead read = readBuilder.newRead().executeFilter();
- RecordReader<InternalRow> reader = read.createReader(splits);
- reader.forEachRemaining(System.out::println);
- }
-}
-```
-
-### Adjust Read Batch Size at Runtime
-
-Parquet and ORC readers can share a `ReadBatchSizer` to adjust the row count
and vector
-capacity of future physical batches without recreating readers:
-
-```java
-import org.apache.paimon.reader.ReadBatchSizer;
-import org.apache.paimon.table.source.TableRead;
-
-ReadBatchSizer sizer = new ReadBatchSizer();
-TableRead read = readBuilder.newRead().withReadBatchSizer(sizer);
-RecordReader<InternalRow> reader = read.createReader(splits);
-
-sizer.setBatchSize(256);
-```
-
-Configure the sizer on `TableRead` before creating readers. A newly created
sizer has no batch size,
-so readers initially use their configured default. Every explicitly set batch
size must be positive.
-Call `clearBatchSize()` to make future physical batches use the configured
default again.
-
-A supporting reader snapshots the batch size before starting a physical batch.
If the size has
-changed, it replaces an idle reusable batch with vectors sized for the new
value and then starts the
-read. The allocation is reused until the batch size changes again.
Consequently, lowering the
-batch size reduces the vector capacity of future batches instead of only
changing their logical
-row count.
-
-An update never mutates a batch that has already started or is still owned by
a consumer.
-Asynchronously prefetched batches may therefore retain the previous size. For
a pooled ORC reader,
-each idle pool entry adopts the current size the next time it is acquired,
while in-flight entries
-keep their old vectors until released. During such a transition, old and new
vectors can coexist.
-
-The sizer uses latest-value semantics: when updates happen faster than
physical batches start,
-readers may skip intermediate batch sizes. Engines should avoid changing the
size too frequently
-because each observed size change reallocates vectors and can add allocation
and garbage-collection
-overhead. A hysteresis interval or minimum adjustment period is recommended.
-
-For concurrent scans, estimate memory using the selected batch size multiplied
by the number of
-active and prefetched batches. This allows an engine to reduce future batch
capacities under memory
-pressure and grow them again when more memory is available.
-
-## Batch Write
-
-The writing is divided into two stages:
-
-1. Write records: Write records in distributed tasks, generate commit messages.
-2. Commit/Abort: Collect all CommitMessages, commit them in a global node
('Coordinator', or named 'Driver', or named 'Committer').
- When the commit fails for certain reason, abort unsuccessful commit via
CommitMessages.
-
-```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.BatchTableCommit;
-import org.apache.paimon.table.sink.BatchTableWrite;
-import org.apache.paimon.table.sink.BatchWriteBuilder;
-import org.apache.paimon.table.sink.CommitMessage;
-
-import java.util.List;
-
-public class BatchWrite {
- public static void main(String[] args) throws Exception {
- // 1. Create a WriteBuilder (Serializable)
- Table table = GetTable.getTable();
- BatchWriteBuilder writeBuilder =
table.newBatchWriteBuilder().withOverwrite();
-
- // 2. Write records in distributed tasks
- BatchTableWrite write = writeBuilder.newWrite();
-
- 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();
-
- // 3. Collect all CommitMessages to a global node and commit
- BatchTableCommit commit = writeBuilder.newCommit();
- commit.commit(messages);
-
- // Abort unsuccessful commit to delete data files
- // commit.abort(messages);
- }
+try (Catalog catalog = CreateCatalog.createFilesystemCatalog()) {
+ Table table = catalog.getTable(Identifier.create("my_db", "my_table"));
+ // Create read or write builders here.
}
```
-## Stream Read
+Next, run the [batch write example](java-writing#batch-write), followed by the
+[batch read example](java-reading#batch-read). Catalog operations and examples
that show a complete
+class declare `throws Exception` for brevity; applications should handle
failures at their own
+error-handling boundary.
-The difference of Stream Read is that StreamTableScan can continuously scan
and generate splits.
+## Read and write data
-StreamTableScan provides the ability to checkpoint and restore, which can let
you save the correct state
-during stream reading.
+### Batch Read
-```java
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateBuilder;
-import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.Split;
-import org.apache.paimon.table.source.StreamTableScan;
-import org.apache.paimon.table.source.TableRead;
-import org.apache.paimon.types.DataTypes;
-import org.apache.paimon.types.RowType;
+See [Java Reads: Batch Read](java-reading#batch-read) for split planning,
projection, and row filtering.
-import com.google.common.collect.Lists;
+#### Adjust Read Batch Size at Runtime
-import java.util.List;
+See [runtime batch sizing](java-reading#adjust-read-batch-size-at-runtime) for
Parquet and ORC readers.
-public class StreamReadTable {
+### Batch Write
- public static void main(String[] args) throws Exception {
- // 1. Create a ReadBuilder and push filter (`withFilter`)
- // and projection (`withProjection`) if necessary
- Table table = GetTable.getTable();
-
- PredicateBuilder builder =
- new PredicateBuilder(RowType.of(DataTypes.STRING(),
DataTypes.INT()));
- Predicate notNull = builder.isNotNull(0);
- Predicate greaterOrEqual = builder.greaterOrEqual(1, 12);
-
- int[] projection = new int[]{0, 1};
-
- ReadBuilder readBuilder =
- table.newReadBuilder()
- .withProjection(projection)
- .withFilter(Lists.newArrayList(notNull,
greaterOrEqual));
-
- // 2. Plan splits in 'Coordinator' (or named 'Driver')
- StreamTableScan scan = readBuilder.newStreamScan();
- while (true) {
- List<Split> splits = scan.plan().splits();
- // Distribute these splits to different tasks
-
- Long state = scan.checkpoint();
- // can be restored in scan.restore(state) after fail over
-
- // 3. Read a split in task
- // You can use executeFilter to do filter per record.
- // By default, only capable of performing coarse-grained filtering.
- TableRead read = readBuilder.newRead().executeFilter();
- RecordReader<InternalRow> reader = read.createReader(splits);
- reader.forEachRemaining(System.out::println);
-
- Thread.sleep(1000);
- }
- }
-}
-```
+See [Java Writes: Batch Write](java-writing#batch-write) for preparing and
committing a batch.
-## Stream Write
+### Stream Read
-The difference of Stream Write is that StreamTableCommit can continuously
commit.
+See [Java Reads: Stream Read](java-reading#stream-read) for continuous
planning and checkpoint recovery.
-Key points to achieve exactly-once consistency:
+### Stream Write
-- CommitUser represents a user. A user can commit multiple times. In
distributed processing, you are
- expected to use the same commitUser.
-- Different applications need to use different commitUsers.
-- The commitIdentifier of `StreamTableWrite` and `StreamTableCommit` needs to
be consistent, and the
- id needs to be incremented for the next committing.
-- When a failure occurs, if you still have uncommitted `CommitMessage`s,
please use `StreamTableCommit#filterAndCommit`
- to exclude the committed messages by commitIdentifier.
+See [Java Writes: Stream Write](java-writing#stream-write) for commit
identities and retry handling.
-```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;
+## Reference
-import java.util.List;
+### Data Types
-public class StreamWriteTable {
+See the [internal value mapping](java-types#data-types) before constructing
`InternalRow` values.
- 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);
- }
- }
-}
-```
+### Predicate Types
-## Data Types
-
-| Java | Paimon |
-|:-------------|:-------------------------------------|
-| boolean | boolean |
-| byte | byte |
-| short | short |
-| int | int |
-| long | long |
-| float | float |
-| double | double |
-| string | org.apache.paimon.data.BinaryString |
-| decimal | org.apache.paimon.data.Decimal |
-| timestamp | org.apache.paimon.data.Timestamp |
-| byte[] | byte[] |
-| array | org.apache.paimon.data.InternalArray |
-| map | org.apache.paimon.data.InternalMap |
-| InternalRow | org.apache.paimon.data.InternalRow |
-
-## Predicate Types
-
-| SQL Predicate | Paimon Predicate
|
-|:--------------|:------------------------------------------------------------|
-| and | org.apache.paimon.predicate.PredicateBuilder.and |
-| or | org.apache.paimon.predicate.PredicateBuilder.or |
-| is null | org.apache.paimon.predicate.PredicateBuilder.isNull |
-| is not null | org.apache.paimon.predicate.PredicateBuilder.isNotNull |
-| in | org.apache.paimon.predicate.PredicateBuilder.in |
-| not in | org.apache.paimon.predicate.PredicateBuilder.notIn |
-| = | org.apache.paimon.predicate.PredicateBuilder.equal |
-| \<> | org.apache.paimon.predicate.PredicateBuilder.notEqual
|
-| \< | org.apache.paimon.predicate.PredicateBuilder.lessThan
|
-| \<= | org.apache.paimon.predicate.PredicateBuilder.lessOrEqual
|
-| > | org.apache.paimon.predicate.PredicateBuilder.greaterThan |
-| >= | org.apache.paimon.predicate.PredicateBuilder.greaterOrEqual |
-| between | org.apache.paimon.predicate.PredicateBuilder.between |
-| like | org.apache.paimon.predicate.PredicateBuilder.like |
-| array contains | org.apache.paimon.predicate.PredicateBuilder.arrayContains |
+See the [predicate reference](java-types#predicate-types) for supported filter
builders.
diff --git a/docs/docs/program-api/java-reading.md
b/docs/docs/program-api/java-reading.md
new file mode 100644
index 0000000000..c3cedcd84f
--- /dev/null
+++ b/docs/docs/program-api/java-reading.md
@@ -0,0 +1,205 @@
+---
+title: "Java Reads"
+sidebar_position: 3
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Java Reads
+
+A read has two stages: a scan plans **splits**, and readers consume those
splits. A standalone
+application can do both. A distributed engine plans centrally and assigns each
split to a reader task.
+
+The examples use `CreateCatalog` and `my_db.my_table` from the [Java API
setup](java-api).
+Populate the table with the [batch write example](java-writing#batch-write)
first.
+
+
+
+## Batch Read
+
+Build predicates against the table's full row type. Filter field indexes refer
to that schema;
+projection selects the columns returned by the reader. This example returns
`f0` and `f1` for
+records whose `f0` is not null and whose `f1` is at least 12.
+
+```java
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class ReadTable {
+
+ public static void main(String[] args) throws Exception {
+ try (Catalog catalog = CreateCatalog.createFilesystemCatalog()) {
+ Table table = catalog.getTable(Identifier.create("my_db",
"my_table"));
+ PredicateBuilder predicates = new
PredicateBuilder(table.rowType());
+ ReadBuilder readBuilder = table.newReadBuilder()
+ .withFilter(Arrays.asList(
+ predicates.isNotNull(0),
predicates.greaterOrEqual(1, 12)))
+ .withProjection(new int[] {0, 1});
+
+ // Plan once in the coordinator; assign splits to workers if
distributed.
+ List<Split> splits = readBuilder.newScan().plan().splits();
+ TableRead read = readBuilder.newRead().executeFilter();
+ try (RecordReader<InternalRow> reader = read.createReader(splits))
{
+ reader.forEachRemaining(row ->
+ System.out.println(row.getString(0) + ", " +
row.getInt(1)));
+ }
+ }
+ }
+}
+```
+
+After the batch write example, the result contains `Alice, 12` and `Emily,
18`; row order is not
+guaranteed. `withFilter` enables pruning, which can leave nonmatching rows in
candidate files.
+`executeFilter()` also evaluates the filter on individual records.
+
+## Reader resources and row reuse
+
+Close readers even if you stop before reaching the end. `forEachRemaining`
consumes and releases
+batches and closes the reader. If you use `readBatch()` directly, call
`releaseBatch()` for every
+returned batch and close the reader when done.
+
+Readers can reuse rows and their backing memory. Consume values before
advancing, or copy the
+values you need to retain. See [Types and Predicates](java-types) for internal
value representations.
+
+## Stream Read
+
+`newStreamScan()` plans successive snapshots. Startup options determine the
initial scan;
+subsequent calls discover new changes. See [streaming
reads](../primary-key-table/table-mode)
+and [scan configuration](../maintenance/configurations) when selecting a
table's read mode.
+
+This example is a single-process polling loop. It prints rows and their
`RowKind`; it is not a
+complete checkpointed pipeline.
+
+```java
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamTableScan;
+import org.apache.paimon.table.source.TableRead;
+
+import java.util.List;
+
+public class StreamReadTable {
+
+ public static void main(String[] args) throws Exception {
+ try (Catalog catalog = CreateCatalog.createFilesystemCatalog()) {
+ Table table = catalog.getTable(Identifier.create("my_db",
"my_table"));
+ ReadBuilder readBuilder = table.newReadBuilder();
+ StreamTableScan scan = readBuilder.newStreamScan();
+ // On recovery, call scan.restore(savedNextSnapshotId) before
planning.
+ TableRead read = readBuilder.newRead();
+
+ while (!Thread.currentThread().isInterrupted()) {
+ List<Split> splits = scan.plan().splits();
+ if (!splits.isEmpty()) {
+ try (RecordReader<InternalRow> reader =
read.createReader(splits)) {
+ reader.forEachRemaining(row -> System.out.println(
+ row.getRowKind() + ": " + row.getString(0) +
", " + row.getInt(1)));
+ }
+ }
+
+ Long nextSnapshotId = scan.checkpoint();
+ // Persist this position together with durable downstream
progress.
+ // Printing to stdout above is not a durable checkpoint.
+ Thread.sleep(1000);
+ }
+ }
+ }
+}
+```
+
+### Checkpoint and restore
+
+`scan.checkpoint()` returns the **next snapshot ID to plan**. It does not
contain the planned
+splits, reader offsets, or downstream state.
+
+- In a sequential reader, finish the planned work and coordinate its durable
downstream result
+ with the saved scan position.
+- In a distributed reader, checkpoint pending splits and reader progress as
well as the scan
+ position. Restoring only the scan position can skip work already planned but
not yet read.
+- Restore the saved position with `scan.restore(nextSnapshotId)` before
planning again. Notify the
+ scan with `notifyCheckpointComplete(nextSnapshotId)` only after the
corresponding checkpoint
+ has completed, when integrating with checkpoint/consumer tracking.
+
+Choose [snapshot retention](../maintenance/manage-snapshots) that leaves
enough time for consumers
+to read and recover. The scan API alone does not provide end-to-end
exactly-once processing.
+
+## Adjust Read Batch Size at Runtime
+
+Parquet and ORC readers can share a `ReadBatchSizer` to adjust the row count
and vector
+capacity of future physical batches without recreating readers. This fragment
uses the
+`readBuilder` and `splits` created in the batch read example:
+
+```java
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.reader.ReadBatchSizer;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.source.TableRead;
+
+ReadBatchSizer sizer = new ReadBatchSizer();
+TableRead read = readBuilder.newRead().withReadBatchSizer(sizer);
+try (RecordReader<InternalRow> reader = read.createReader(splits)) {
+ sizer.setBatchSize(256);
+ // Consume batches with reader.readBatch() and release each batch after
use.
+}
+```
+
+Configure the sizer on `TableRead` before creating readers. A newly created
sizer has no batch size,
+so readers initially use their configured default. Every explicitly set batch
size must be positive.
+Call `clearBatchSize()` to make future physical batches use the configured
default again.
+
+A supporting reader snapshots the batch size before starting a physical batch.
If the size has
+changed, it replaces an idle reusable batch with vectors sized for the new
value and then starts the
+read. The allocation is reused until the batch size changes again.
Consequently, lowering the
+batch size reduces the vector capacity of future batches instead of only
changing their logical
+row count.
+
+An update never mutates a batch that has already started or is still owned by
a consumer.
+Asynchronously prefetched batches may therefore retain the previous size. For
a pooled ORC reader,
+each idle pool entry adopts the current size the next time it is acquired,
while in-flight entries
+keep their old vectors until released. During such a transition, old and new
vectors can coexist.
+
+The sizer uses latest-value semantics: when updates happen faster than
physical batches start,
+readers may skip intermediate batch sizes. Engines should avoid changing the
size too frequently
+because each observed size change reallocates vectors and can add allocation
and garbage-collection
+overhead. A hysteresis interval or minimum adjustment period is recommended.
+
+For concurrent scans, estimate memory using the selected batch size multiplied
by the number of
+active and prefetched batches. This allows an engine to reduce future batch
capacities under memory
+pressure and grow them again when more memory is available.
+
+## Next steps
+
+Use [Java Writes](java-writing) to publish data, [Types and
Predicates](java-types) to construct
+filters, and [Local Cache](file-cache) to reduce repeated file reads.
diff --git a/docs/docs/program-api/java-types.md
b/docs/docs/program-api/java-types.md
new file mode 100644
index 0000000000..b36c0b98e0
--- /dev/null
+++ b/docs/docs/program-api/java-types.md
@@ -0,0 +1,96 @@
+---
+title: "Types and Predicates"
+sidebar_position: 5
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Types and Predicates
+
+The Java table API exchanges `InternalRow` values. Use Paimon's internal value
representations
+when constructing rows or predicate literals; SQL type names do not imply that
arbitrary Java
+objects can be placed in a row.
+
+## Data Types
+
+| Paimon logical type | Internal Java value |
+| --- | --- |
+| `BOOLEAN` | `boolean` / `Boolean` |
+| `TINYINT` | `byte` / `Byte` |
+| `SMALLINT` | `short` / `Short` |
+| `INT` | `int` / `Integer` |
+| `BIGINT` | `long` / `Long` |
+| `FLOAT` | `float` / `Float` |
+| `DOUBLE` | `double` / `Double` |
+| `CHAR`, `VARCHAR`, `STRING` | `org.apache.paimon.data.BinaryString` |
+| `DECIMAL` | `org.apache.paimon.data.Decimal` |
+| `DATE` | `int`, days since the Unix epoch |
+| `TIME` | `int`, milliseconds since midnight |
+| `TIMESTAMP`, `TIMESTAMP_LTZ` | `org.apache.paimon.data.Timestamp` |
+| `BINARY`, `VARBINARY`, `BYTES` | `byte[]` |
+| `ARRAY` | `org.apache.paimon.data.InternalArray` |
+| `MAP` | `org.apache.paimon.data.InternalMap` |
+| `ROW` | `org.apache.paimon.data.InternalRow` |
+
+For example, the sample schema `(f0 STRING, f1 INT)` accepts:
+
+```java
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+
+GenericRow row = GenericRow.of(BinaryString.fromString("Alice"), 12);
+```
+
+`GenericRow` uses insert row kind by default. For changelog writes, use
Paimon's
+`org.apache.paimon.types.RowKind` and the table's supported change semantics.
Flink's external
+`Row` and `RowKind` are separate types; the [Flink API](flink-api) converts
them.
+See [Data Types](../concepts/data-types) for logical type definitions and
additional types.
+
+## Predicate Types
+
+Construct a `PredicateBuilder` from `table.rowType()`. Field indexes refer to
the original table
+schema, even when the read uses projection.
+
+| SQL predicate | `PredicateBuilder` method |
+| --- | --- |
+| `AND`, `OR` | `and`, `or` |
+| `IS NULL`, `IS NOT NULL` | `isNull`, `isNotNull` |
+| `IN`, `NOT IN` | `in`, `notIn` |
+| `=`, `<>` | `equal`, `notEqual` |
+| `<`, `<=` | `lessThan`, `lessOrEqual` |
+| `>`, `>=` | `greaterThan`, `greaterOrEqual` |
+| `BETWEEN` | `between` |
+| `LIKE` | `like` |
+| Array membership | `arrayContains` |
+
+```java
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+
+PredicateBuilder builder = new PredicateBuilder(table.rowType());
+Predicate filter = PredicateBuilder.and(
+ builder.equal(0, BinaryString.fromString("Alice")),
+ builder.greaterOrEqual(1, 12));
+```
+
+Pass predicates to `ReadBuilder.withFilter`. Enable
`TableRead.executeFilter()` when the reader
+must also evaluate them per row; pruning alone can return candidate rows that
do not match.
+See [Java Reads](java-reading#batch-read) for a complete example.
diff --git a/docs/docs/program-api/java-writing.md
b/docs/docs/program-api/java-writing.md
new file mode 100644
index 0000000000..3d29235d73
--- /dev/null
+++ b/docs/docs/program-api/java-writing.md
@@ -0,0 +1,182 @@
+---
+title: "Java Writes"
+sidebar_position: 4
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Java Writes
+
+Writing records creates files; committing makes the file changes visible
through table snapshots.
+A distributed job writes in worker tasks, gathers their `CommitMessage`s, and
commits centrally.
+
+The examples use `CreateCatalog` and the two-column primary-key table from the
+[Java API setup](java-api). Java strings must be converted to `BinaryString`
before writing.
+
+
+
+## Batch Write
+
+Create a `BatchWriteBuilder`, write records, prepare messages, and commit the
collected messages
+once. This example upserts three rows into the sample primary-key table.
+
+```java
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+
+import java.util.List;
+
+public class BatchWrite {
+
+ public static void main(String[] args) throws Exception {
+ try (Catalog catalog = CreateCatalog.createFilesystemCatalog()) {
+ Table table = catalog.getTable(Identifier.create("my_db",
"my_table"));
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(BinaryString.fromString("Alice"),
12));
+ write.write(GenericRow.of(BinaryString.fromString("Bob"), 5));
+ write.write(GenericRow.of(BinaryString.fromString("Emily"),
18));
+
+ List<CommitMessage> messages = write.prepareCommit();
+ // In a distributed job, collect messages from every writer
first.
+ commit.commit(messages);
+ }
+ }
+ }
+}
+```
+
+Closing a writer does not publish its data. Call `prepareCommit()` and commit
the resulting
+messages before closing it. A batch committer is for one commit; use a new
builder for another batch.
+
+### Overwrite and abort
+
+Use `table.newBatchWriteBuilder().withOverwrite()` only when you intend to
replace existing data.
+For a partitioned table, check the overwrite scope and table options before
selecting this mode;
+see [overwrite behavior](../flink/sql-write#overwriting-the-whole-table).
+
+`BatchTableCommit.abort(messages)` deletes files belonging to an abandoned
write. Use it only
+for messages that are known not to have been committed and will not be
retried. A commit exception
+alone does not prove that publishing failed; do not delete files from a
possibly successful commit.
+
+## Stream Write
+
+A `StreamTableWrite` and `StreamTableCommit` can be reused across commits. The
application supplies
+two identities:
+
+| Identity | Rule |
+| --- | --- |
+| `commitUser` | Shared by writers and committer of one application, stable
across recovery, different for independent applications. |
+| `commitIdentifier` | The same value for prepare and commit, increasing for
subsequent commits. A checkpoint ID is a common choice. |
+
+The default commit user is random. Set it explicitly, or persist and restore
the generated value,
+when implementing recovery.
+
+```java
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+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 {
+ // Supply a unique identity for a new application; retain it on
recovery.
+ String commitUser = args[0];
+ try (Catalog catalog = CreateCatalog.createFilesystemCatalog()) {
+ Table table = catalog.getTable(Identifier.create("my_db",
"my_table"));
+ StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder()
+ .withCommitUser(commitUser);
+ try (StreamTableWrite write = writeBuilder.newWrite();
+ StreamTableCommit commit = writeBuilder.newCommit()) {
+ // This finite example starts a NEW application without
recovered state.
+ for (long commitIdentifier = 0; commitIdentifier < 3;
commitIdentifier++) {
+ write.write(GenericRow.of(
+ BinaryString.fromString("Alice"), 12 + (int)
commitIdentifier));
+ List<CommitMessage> messages =
+ write.prepareCommit(false, commitIdentifier);
+
+ // A recoverable pipeline saves messages and input
progress durably here.
+ commit.commit(commitIdentifier, messages);
+ // The loop advances the ID only after this commit
succeeds.
+ }
+ }
+ }
+ }
+}
+```
+
+Pass an application identity as the first argument. This finite example
demonstrates matching
+prepare/commit IDs; it does not persist checkpoints. Do not restart it from ID
0 with a previously
+used identity and treat it as recovery.
+
+### Recover a pending commit
+
+Save the commit user, commit IDs, prepared messages, and associated input
progress in your
+runtime's checkpoint state. After restoring the same commit user, retry saved
messages with
+`filterAndCommit` if the outcome of an earlier commit is uncertain:
+
+```java
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+// restoredCommitIdentifier and restoredMessages come from durable checkpoint
state.
+Map<Long, List<CommitMessage>> pending = new HashMap<>();
+pending.put(restoredCommitIdentifier, restoredMessages);
+try (StreamTableCommit commit = writeBuilder.newCommit()) {
+ commit.filterAndCommit(pending);
+}
+```
+
+`commit` does not check whether an ID was already committed. `filterAndCommit`
filters previously
+committed IDs and publishes the remaining messages. Resume new writes with the
next ID after
+reconciling pending commits. This protocol must be coordinated with source
replay and writer
+recovery to provide end-to-end exactly-once behavior.
+
+## Distributed writer routing
+
+Create one logical write builder for the operation and use it for writers and
the committer.
+For the fixed-bucket table used here, use `writeBuilder.newWriteSelector()` to
determine which
+downstream writer should receive a record. Records for the same bucket must be
routed consistently.
+Bucket-unaware and postpone modes return no selector. Dynamic-bucket modes do
not support this
+selector API: they require dedicated bucket assignment and `write(row,
bucket)` instead. See
+[data distribution](../primary-key-table/data-distribution) when selecting a
different table layout.
+
+For a Flink job, use [FlinkSinkBuilder](flink-api#write-to-table) to integrate
routing, checkpoints,
+and commits with the engine.
diff --git a/docs/docs/program-api/rest-api.mdx
b/docs/docs/program-api/rest-api.mdx
index 450636b215..af6d79786f 100644
--- a/docs/docs/program-api/rest-api.mdx
+++ b/docs/docs/program-api/rest-api.mdx
@@ -1,6 +1,6 @@
---
-title: "REST API"
-sidebar_position: 1
+title: "REST Java Client"
+sidebar_position: 7
---
import Stable from '@site/src/components/Stable';
@@ -25,9 +25,17 @@ specific language governing permissions and limitations
under the License.
-->
-# REST API
+# REST Java Client
-This is Java API for [REST](../concepts/rest/).
+`RESTApi` is a Java client for the [Paimon REST catalog](../concepts/rest/).
Use it to issue catalog
+metadata requests without bringing in the full table read/write bundle.
+
+| What you need | API or reference |
+| --- | --- |
+| List and manage catalog objects from Java | `RESTApi`, shown below |
+| Load a `Table` and read or write rows | [Java API](java-api) with a REST
catalog |
+| Implement an HTTP client or catalog server | [REST API
specification](../concepts/rest/rest-api) |
+| Administrative endpoints | [Management API](../concepts/rest/management-api)
|
## Dependency
@@ -53,7 +61,11 @@ Or download the jar file:
</Unstable>
-## RESTApi
+## Connect and list tables {#restapi}
+
+Set the server URI, warehouse identifier, and authentication options. The
warehouse value is
+interpreted by the server; it is not necessarily a filesystem path. The
example uses the bearer
+token provider, whose configuration value is spelled `bear`.
```java
import org.apache.paimon.options.Options;
@@ -74,14 +86,14 @@ public class RESTApiExample {
Options options = new Options();
options.set(URI, "<catalog server url>");
options.set(WAREHOUSE, "my_instance_name");
- setBearToken(options); // or setDlfToken
+ setBearerToken(options); // or setDlfToken
RESTApi api = new RESTApi(options);
List<String> tables = api.listTables("my_database");
System.out.println(tables);
}
- private static void setBearToken(Options options) {
+ private static void setBearerToken(Options options) {
options.set(TOKEN_PROVIDER, "bear");
options.set(TOKEN, "<token>");
}
@@ -94,4 +106,13 @@ public class RESTApiExample {
}
```
-See more methods in `'RESTApi'`.
+## Authentication and table access
+
+See [bearer authentication](../concepts/rest/bear) or [DLF
authentication](../concepts/rest/dlf)
+for provider configuration. Supply credentials through your application's
configuration; the
+placeholders above illustrate the option names.
+
+To use the full table API, add the [Java bundle](java-api#dependency) and
create a catalog with
+`metastore=rest`, the server `uri`, `warehouse`, and the same authentication
options. See
+[REST catalog configuration](../concepts/rest/) for an example. Loading table
metadata and reading
+its data files are separate operations, so configure storage access as
required by the catalog.
diff --git a/docs/docs/program-api/rust-api.md
b/docs/docs/program-api/rust-api.md
index c34aa46107..7a8e4b6263 100644
--- a/docs/docs/program-api/rust-api.md
+++ b/docs/docs/program-api/rust-api.md
@@ -1,9 +1,8 @@
---
title: "Rust API"
-sidebar_position: 7
+sidebar_position: 9
---
-
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
@@ -25,5 +24,23 @@ under the License.
# Rust API
-- [Paimon Rust Docs](https://paimon.apache.org/docs/rust/).
-- [Paimon Rust Repo](https://github.com/apache/paimon-rust).
+[Paimon Rust](https://github.com/apache/paimon-rust) provides native access to
the Paimon table format
+from Rust applications. It has its own release cycle and dependency
configuration.
+
+## Get started
+
+Follow the [Rust getting-started
guide](https://paimon.apache.org/docs/rust/getting-started/) to add
+the crate, select storage features, create a catalog, and read a table. Use
the crate version and
+feature flags from that guide for your chosen release.
+
+## Choose a guide
+
+| Task | Documentation |
+| --- | --- |
+| Install the crate and read a table | [Getting
Started](https://paimon.apache.org/docs/rust/getting-started/) |
+| Explore supported features and integrations | [Rust
Documentation](https://paimon.apache.org/docs/rust/) |
+| Build from source or contribute | [Paimon Rust
Repository](https://github.com/apache/paimon-rust) |
+
+Catalog and table concepts are shared across implementations. See
[Catalogs](../concepts/catalog)
+and the [Storage Specification](../concepts/spec/) for the format, then check
the Rust documentation
+for the APIs and features supported by the version you use.
diff --git a/docs/sidebars.js b/docs/sidebars.js
index 67670e7ec7..099fb7f0b7 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -591,12 +591,27 @@ const sidebars = {
"id": "program-api/index"
},
"items": [
- "program-api/rest-api",
+ {
+ type: "category",
+ "label": "Java API",
+ "link": {type: "doc", "id": "program-api/java-api"},
+ "items": [
+ "program-api/catalog-api",
+ "program-api/java-reading",
+ "program-api/java-writing",
+ "program-api/java-types"
+ ]
+ },
"program-api/flink-api",
- "program-api/java-api",
- "program-api/catalog-api",
- "program-api/cpp-api",
- "program-api/rust-api",
+ {
+ type: "category",
+ "label": "Other Clients",
+ "items": [
+ "program-api/rest-api",
+ "program-api/cpp-api",
+ "program-api/rust-api"
+ ]
+ },
"program-api/file-cache"
]
},
diff --git a/docs/static/img/program-api-local-cache.svg
b/docs/static/img/program-api-local-cache.svg
new file mode 100644
index 0000000000..6d59b627a9
--- /dev/null
+++ b/docs/static/img/program-api-local-cache.svg
@@ -0,0 +1,54 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="960" height="586" viewBox="0 0
960 586" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Local blocks reduce repeated storage reads</title>
+<desc id="desc">Eligible immutable file reads check the local memory or disk
cache. Hits return a cached block. Misses fetch and cache a block from storage.
Non-whitelisted files and mutable hints bypass the cache.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="958" height="584" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="44" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Local blocks reduce repeated storage reads</text>
+<text x="28" y="76" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Enable caching on the catalog and choose which file types
to cache.</text>
+<rect x="28" y="116" width="238" height="104" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="147.0" y="149" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">File read</text>
+<text x="147.0" y="177" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Check file type</text>
+<text x="147.0" y="202" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">and mutable hints</text>
+<rect x="365" y="116" width="257" height="104" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="493.5" y="149" font-size="21" fill="#087f6e" font-weight="700"
text-anchor="middle">Local block cache</text>
+<text x="493.5" y="177" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Memory or disk</text>
+<text x="493.5" y="202" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Bounded by max-size</text>
+<rect x="720" y="116" width="212" height="104" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="826.0" y="149" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="middle">Storage</text>
+<text x="826.0" y="177" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">S3 / OSS / HDFS</text>
+<text x="826.0" y="202" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">or filesystem</text>
+<path d="M 266 165 H 365" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<text x="315" y="142" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Eligible</text>
+<path d="M 622 165 H 720" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<text x="671" y="142" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Miss</text>
+<path d="M 824 220 V 265 H 560 V 220" fill="none" stroke="#526277"
stroke-width="2" marker-end="url(#arrow)"/>
+<text x="744" y="292" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Fetch block and cache it</text>
+<text x="28" y="332" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">Other file types, LATEST and</text>
+<text x="28" y="357" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">EARLIEST read storage directly.</text>
+<rect x="365" y="390" width="257" height="80" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="493.5" y="423" font-size="21" fill="#087f6e" font-weight="700"
text-anchor="middle">Return bytes</text>
+<text x="493.5" y="451" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Hit: reuse cached block</text>
+<path d="M 493 220 V 390" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<text x="28" y="517" font-size="18" fill="#172b4d" font-weight="700"
text-anchor="start">Default whitelist: meta,global-index. Add data explicitly
to cache data files.</text>
+<text x="28" y="548" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">Block hits avoid content downloads; file-status and other
metadata calls can still occur.</text>
+</g>
+</svg>
diff --git a/docs/static/img/program-api-overview.svg
b/docs/static/img/program-api-overview.svg
new file mode 100644
index 0000000000..b017468323
--- /dev/null
+++ b/docs/static/img/program-api-overview.svg
@@ -0,0 +1,63 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="960" height="560" viewBox="0 0
960 560" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Choose the layer your application needs</title>
+<desc id="desc">Catalog clients manage metadata. Table APIs plan reads and
prepare writes. Flink and Spark integrate table access with execution and
recovery.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="958" height="558" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="44" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Choose the layer your application needs</text>
+<text x="28" y="76" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Catalogs resolve tables; table APIs expose data access;
engines coordinate execution.</text>
+<rect x="28" y="108" width="275" height="104" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="165.5" y="141" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">Catalog clients</text>
+<text x="165.5" y="169" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Java Catalog / RESTApi</text>
+<text x="165.5" y="194" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">C++ / Rust catalogs</text>
+<rect x="343" y="108" width="275" height="104" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="480.5" y="141" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">Table APIs</text>
+<text x="480.5" y="169" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Java / C++ / Rust / Python</text>
+<text x="480.5" y="194" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Rows, splits, commit messages</text>
+<rect x="658" y="108" width="274" height="104" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="795.0" y="141" font-size="21" fill="#087f6e" font-weight="700"
text-anchor="middle">Engine integrations</text>
+<text x="795.0" y="169" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Flink / Spark</text>
+<text x="795.0" y="194" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Jobs and checkpoints</text>
+<path d="M 303 160 H 343" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 658 160 H 618" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="28" y="262" width="275" height="109" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="165.5" y="295" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="middle">Catalog metadata</text>
+<text x="165.5" y="323" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Databases and table schemas</text>
+<text x="165.5" y="348" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Names, locations, options</text>
+<rect x="343" y="262" width="275" height="109" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="480.5" y="295" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="middle">Read and write builders</text>
+<text x="480.5" y="323" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Plan splits → read rows</text>
+<text x="480.5" y="348" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Write files → prepare changes</text>
+<rect x="658" y="262" width="274" height="109" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="795.0" y="295" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="middle">Runtime coordination</text>
+<text x="795.0" y="323" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Assign splits and route rows</text>
+<text x="795.0" y="348" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Commit and recover</text>
+<path d="M 165 212 V 262" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 480 212 V 262" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 795 212 V 262" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="28" y="422" width="904" height="99" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="480" y="458" font-size="23" fill="#087f6e" font-weight="700"
text-anchor="middle">Paimon table storage</text>
+<text x="480" y="488" font-size="18" fill="#526277" font-weight="400"
text-anchor="middle">Schemas • snapshots • manifests • data files •
indexes</text>
+<path d="M 165 371 V 422" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 480 371 V 422" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 795 371 V 422" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+</g>
+</svg>
diff --git a/docs/static/img/program-api-read-flow.svg
b/docs/static/img/program-api-read-flow.svg
new file mode 100644
index 0000000000..47c63f4906
--- /dev/null
+++ b/docs/static/img/program-api-read-flow.svg
@@ -0,0 +1,52 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="960" height="534" viewBox="0 0
960 534" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Plan once, read the assigned splits</title>
+<desc id="desc">A coordinator configures a ReadBuilder and plans splits.
Worker readers use the same read configuration. A streaming checkpoint must
also account for pending splits and reader progress.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="958" height="532" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="44" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Plan once, read the assigned splits</text>
+<text x="28" y="76" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Use the same filter and projection when planning and
reading.</text>
+<rect x="28" y="110" width="282" height="118" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="169.0" y="143" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">Coordinator</text>
+<text x="169.0" y="171" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">ReadBuilder</text>
+<text x="169.0" y="196" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">scan.plan().splits()</text>
+<rect x="385" y="110" width="200" height="118" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="485.0" y="143" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="middle">Splits</text>
+<text x="485.0" y="171" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Split A</text>
+<text x="485.0" y="196" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Split B</text>
+<rect x="660" y="110" width="272" height="93" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="796.0" y="143" font-size="21" fill="#087f6e" font-weight="700"
text-anchor="middle">Reader task 1</text>
+<text x="796.0" y="171" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Read Split A</text>
+<rect x="660" y="235" width="272" height="93" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="796.0" y="268" font-size="21" fill="#087f6e" font-weight="700"
text-anchor="middle">Reader task 2</text>
+<text x="796.0" y="296" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Read Split B</text>
+<path d="M 310 167 H 385" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 585 147 H 660" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 585 192 H 622 V 278 H 660" fill="none" stroke="#526277"
stroke-width="2" marker-end="url(#arrow)"/>
+<text x="28" y="273" font-size="20" fill="#172b4d" font-weight="700"
text-anchor="start">Each reader</text>
+<text x="28" y="305" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">newRead() → createReader(assigned splits)</text>
+<text x="28" y="336" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">executeFilter() adds per-row filtering.</text>
+<text x="28" y="367" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Release batches and close readers after use.</text>
+<rect x="28" y="408" width="904" height="95" rx="8" fill="#fff6e8"
stroke="#c5862b" stroke-width="1.5"/>
+<text x="48" y="440" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="start">Streaming recovery needs more than the scan position</text>
+<text x="48" y="470" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Save next snapshot ID + pending splits + reader and
downstream progress.</text>
+</g>
+</svg>
diff --git a/docs/static/img/program-api-write-flow.svg
b/docs/static/img/program-api-write-flow.svg
new file mode 100644
index 0000000000..e31bea4cf3
--- /dev/null
+++ b/docs/static/img/program-api-write-flow.svg
@@ -0,0 +1,56 @@
+<svg xmlns="http://www.w3.org/2000/svg"; width="960" height="598" viewBox="0 0
960 598" role="img" aria-labelledby="title desc">
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<title id="title">Prepare files, then publish a commit</title>
+<desc id="desc">For streaming commit N, workers write and prepare messages
with N. A coordinator commits all messages with N and the same commit user. A
recovery path restores saved messages and filters already committed IDs.</desc>
+<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7"
markerHeight="7" orient="auto"><path d="M 0 0 L 10 5 L 0 10 z"
fill="#526277"/></marker></defs>
+<g font-family="Arial, Helvetica, sans-serif">
+<rect x="1" y="1" width="958" height="596" rx="12" fill="#ffffff"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="28" y="44" font-size="27" fill="#172b4d" font-weight="700"
text-anchor="start">Prepare files, then publish a commit</text>
+<text x="28" y="76" font-size="18" fill="#526277" font-weight="400"
text-anchor="start">Streaming example: one application identity, one matching
ID for each commit.</text>
+<rect x="28" y="111" width="277" height="110" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="166.5" y="144" font-size="21" fill="#2463b4" font-weight="700"
text-anchor="middle">Worker writers</text>
+<text x="166.5" y="172" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">write(row)</text>
+<text x="166.5" y="197" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">prepareCommit(false, N)</text>
+<rect x="350" y="111" width="242" height="110" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="471.0" y="144" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="middle">Prepared messages</text>
+<text x="471.0" y="172" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Collect from all writers</text>
+<text x="471.0" y="197" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Keep commit ID N</text>
+<rect x="637" y="111" width="295" height="110" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="784.5" y="144" font-size="21" fill="#087f6e" font-weight="700"
text-anchor="middle">Coordinator</text>
+<text x="784.5" y="172" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">commit(N, messages)</text>
+<text x="784.5" y="197" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Publish snapshot changes</text>
+<path d="M 305 166 H 350" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<path d="M 592 166 H 637" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<rect x="28" y="270" width="904" height="64" rx="8" fill="#eaf2ff"
stroke="#2463b4" stroke-width="1.5"/>
+<text x="480" y="308" font-size="19" fill="#2463b4" font-weight="700"
text-anchor="middle">Use the same commitUser and commitIdentifier for prepare
and commit.</text>
+<path d="M 471 221 V 270" fill="none" stroke="#526277" stroke-width="2"
stroke-dasharray="6 5" marker-end="url(#arrow)"/>
+<rect x="28" y="386" width="418" height="111" rx="8" fill="#f5f7fb"
stroke="#d7dfeb" stroke-width="1.5"/>
+<text x="237.0" y="419" font-size="21" fill="#172b4d" font-weight="700"
text-anchor="middle">Recover an uncertain commit</text>
+<text x="237.0" y="447" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Restore saved user, IDs, messages</text>
+<text x="237.0" y="472" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">and associated input progress</text>
+<rect x="514" y="386" width="418" height="111" rx="8" fill="#e8f7f2"
stroke="#087f6e" stroke-width="1.5"/>
+<text x="723.0" y="419" font-size="21" fill="#087f6e" font-weight="700"
text-anchor="middle">Retry saved messages</text>
+<text x="723.0" y="447" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">filterAndCommit(pending)</text>
+<text x="723.0" y="472" font-size="17" fill="#526277" font-weight="400"
text-anchor="middle">Skip IDs already committed</text>
+<path d="M 446 441 H 514" fill="none" stroke="#526277" stroke-width="2"
marker-end="url(#arrow)"/>
+<text x="28" y="545" font-size="19" fill="#172b4d" font-weight="700"
text-anchor="start">Advance the ID for new work after reconciling pending
commits.</text>
+<text x="28" y="574" font-size="17" fill="#526277" font-weight="400"
text-anchor="start">Batch writes follow prepare → commit once, without an
application-supplied commit ID.</text>
+</g>
+</svg>
