This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit b3b5660c38df769269adfa835b9460c010f7fd34 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>
