Java API

We do not recommend using the Paimon API naked, unless you are a professional downstream ecosystem developer, and even if you do, there will be significant difficulties.

If you are only using Paimon, we strongly recommend using computing engines such as Flink SQL or Spark SQL.

The following documents are not detailed and are for reference only.

Dependency

Maven dependency:

  1. <dependency>
  2.   <groupId>org.apache.paimon</groupId>
  3.   <artifactId>paimon-bundle</artifactId>
  4.   <version>0.9.0</version>
  5. </dependency>

Or download the jar file: Paimon Bundle.

Paimon relies on Hadoop environment, you should add hadoop classpath or bundled jar.

Create Catalog

Before coming into contact with the Table, you need to create a Catalog.

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.CatalogContext;
  3. import org.apache.paimon.catalog.CatalogFactory;
  4. import org.apache.paimon.fs.Path;
  5. import org.apache.paimon.options.Options;
  7. public class CreateCatalog {
  9.     public static Catalog createFilesystemCatalog() {
  10.         CatalogContext context = CatalogContext.create(new Path("..."));
  11.         return CatalogFactory.createCatalog(context);
  12.     }
  14.     public static Catalog createHiveCatalog() {
  15.         // Paimon Hive catalog relies on Hive jars
  16.         // You should add hive classpath or hive bundled jar.
  17.         Options options = new Options();
  18.         options.set("warehouse", "...");
  19.         options.set("metastore", "hive");
  20.         options.set("uri", "...");
  21.         options.set("hive-conf-dir", "...");
  22.         options.set("hadoop-conf-dir", "...");
  23.         CatalogContext context = CatalogContext.create(options);
  24.         return CatalogFactory.createCatalog(context);
  25.     }
  26. }

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.

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.Identifier;
  3. import org.apache.paimon.schema.Schema;
  4. import org.apache.paimon.types.DataTypes;
  6. public class CreateTable {
  8.     public static void main(String[] args) {
  9.         Schema.Builder schemaBuilder = Schema.newBuilder();
  10.         schemaBuilder.primaryKey("f0", "f1");
  11.         schemaBuilder.partitionKeys("f1");
  12.         schemaBuilder.column("f0", DataTypes.STRING());
  13.         schemaBuilder.column("f1", DataTypes.INT());
  14.         Schema schema = schemaBuilder.build();
  16.         Identifier identifier = Identifier.create("my_db", "my_table");
  17.         try {
  18.             Catalog catalog = CreateCatalog.createFilesystemCatalog();
  19.             catalog.createTable(identifier, schema, false);
  20.         } catch (Catalog.TableAlreadyExistException e) {
  21.             // do something
  22.         } catch (Catalog.DatabaseNotExistException e) {
  23.             // do something
  24.         }
  25.     }
  26. }

Get Table

The Table interface provides access to the table metadata and tools to read and write table.

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.Identifier;
  3. import org.apache.paimon.table.Table;
  5. public class GetTable {
  7.     public static Table getTable() {
  8.         Identifier identifier = Identifier.create("my_db", "my_table");
  9.         try {
  10.             Catalog catalog = CreateCatalog.createFilesystemCatalog();
  11.             return catalog.getTable(identifier);
  12.         } catch (Catalog.TableNotExistException e) {
  13.             // do something
  14.             throw new RuntimeException("table not exist");
  15.         }
  16.     }
  17. }

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.
  1. import org.apache.paimon.data.InternalRow;
  2. import org.apache.paimon.predicate.Predicate;
  3. import org.apache.paimon.predicate.PredicateBuilder;
  4. import org.apache.paimon.reader.RecordReader;
  5. import org.apache.paimon.table.Table;
  6. import org.apache.paimon.table.source.ReadBuilder;
  7. import org.apache.paimon.table.source.Split;
  8. import org.apache.paimon.table.source.TableRead;
  9. import org.apache.paimon.types.DataTypes;
  10. import org.apache.paimon.types.RowType;
  12. import com.google.common.collect.Lists;
  14. import java.util.List;
  16. public class ReadTable {
  18.     public static void main(String[] args) throws Exception {
  19.         // 1. Create a ReadBuilder and push filter (`withFilter`)
  20.         // and projection (`withProjection`) if necessary
  21.         Table table = GetTable.getTable();
  23.         PredicateBuilder builder =
  24.             new PredicateBuilder(RowType.of(DataTypes.STRING(), DataTypes.INT()));
  25.         Predicate notNull = builder.isNotNull(0);
  26.         Predicate greaterOrEqual = builder.greaterOrEqual(1, 12);
  28.         int[] projection = new int[] {0, 1};
  30.         ReadBuilder readBuilder =
  31.             table.newReadBuilder()
  32.                 .withProjection(projection)
  33.                 .withFilter(Lists.newArrayList(notNull, greaterOrEqual));
  35.         // 2. Plan splits in 'Coordinator' (or named 'Driver')
  36.         List<Split> splits = readBuilder.newScan().plan().splits();
  38.         // 3. Distribute these splits to different tasks
  40.         // 4. Read a split in task
  41.         TableRead read = readBuilder.newRead();
  42.         RecordReader<InternalRow> reader = read.createReader(splits);
  43.         reader.forEachRemaining(System.out::println);
  44.     }
  45. }

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.
  1. import org.apache.paimon.data.BinaryString;
  2. import org.apache.paimon.data.GenericRow;
  3. import org.apache.paimon.table.Table;
  4. import org.apache.paimon.table.sink.BatchTableCommit;
  5. import org.apache.paimon.table.sink.BatchTableWrite;
  6. import org.apache.paimon.table.sink.BatchWriteBuilder;
  7. import org.apache.paimon.table.sink.CommitMessage;
  9. import java.util.List;
  11. public class BatchWrite {
  12.     public static void main(String[] args) throws Exception {
  13.         // 1. Create a WriteBuilder (Serializable)
  14.         Table table = GetTable.getTable();
  15.         BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder().withOverwrite();
  17.         // 2. Write records in distributed tasks
  18.         BatchTableWrite write = writeBuilder.newWrite();
  20.         GenericRow record1 = GenericRow.of(BinaryString.fromString("Alice"), 12);
  21.         GenericRow record2 = GenericRow.of(BinaryString.fromString("Bob"), 5);
  22.         GenericRow record3 = GenericRow.of(BinaryString.fromString("Emily"), 18);
  24.         // If this is a distributed write, you can use writeBuilder.newWriteSelector.
  25.         // WriteSelector determines to which logical downstream writers a record should be written to.
  26.         // If it returns empty, no data distribution is required.
  28.         write.write(record1);
  29.         write.write(record2);
  30.         write.write(record3);
  32.         List<CommitMessage> messages = write.prepareCommit();
  34.         // 3. Collect all CommitMessages to a global node and commit
  35.         BatchTableCommit commit = writeBuilder.newCommit();
  36.         commit.commit(messages);
  38.         // Abort unsuccessful commit to delete data files
  39.         // commit.abort(messages);
  40.     }
  41. }

Stream Read

The difference of Stream Read is that StreamTableScan can continuously scan and generate splits.

StreamTableScan provides the ability to checkpoint and restore, which can let you save the correct state during stream reading.

  1. import org.apache.paimon.data.InternalRow;
  2. import org.apache.paimon.predicate.Predicate;
  3. import org.apache.paimon.predicate.PredicateBuilder;
  4. import org.apache.paimon.reader.RecordReader;
  5. import org.apache.paimon.table.Table;
  6. import org.apache.paimon.table.source.ReadBuilder;
  7. import org.apache.paimon.table.source.Split;
  8. import org.apache.paimon.table.source.StreamTableScan;
  9. import org.apache.paimon.table.source.TableRead;
  10. import org.apache.paimon.types.DataTypes;
  11. import org.apache.paimon.types.RowType;
  13. import com.google.common.collect.Lists;
  15. import java.util.List;
  17. public class StreamReadTable {
  19.     public static void main(String[] args) throws Exception {
  20.         // 1. Create a ReadBuilder and push filter (`withFilter`)
  21.         // and projection (`withProjection`) if necessary
  22.         Table table = GetTable.getTable();
  24.         PredicateBuilder builder =
  25.             new PredicateBuilder(RowType.of(DataTypes.STRING(), DataTypes.INT()));
  26.         Predicate notNull = builder.isNotNull(0);
  27.         Predicate greaterOrEqual = builder.greaterOrEqual(1, 12);
  29.         int[] projection = new int[] {0, 1};
  31.         ReadBuilder readBuilder =
  32.             table.newReadBuilder()
  33.                 .withProjection(projection)
  34.                 .withFilter(Lists.newArrayList(notNull, greaterOrEqual));
  36.         // 2. Plan splits in 'Coordinator' (or named 'Driver')
  37.         StreamTableScan scan = readBuilder.newStreamScan();
  38.         while (true) {
  39.             List<Split> splits = scan.plan().splits();
  40.             // Distribute these splits to different tasks
  42.             Long state = scan.checkpoint();
  43.             // can be restored in scan.restore(state) after fail over
  45.             // 3. Read a split in task
  46.             TableRead read = readBuilder.newRead();
  47.             RecordReader<InternalRow> reader = read.createReader(splits);
  48.             reader.forEachRemaining(System.out::println);
  50.             Thread.sleep(1000);
  51.         }
  52.     }
  53. }

Stream Write

The difference of Stream Write is that StreamTableCommit can continuously commit.

Key points to achieve exactly-once consistency:

  • 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 CommitMessages, please use StreamTableCommit#filterAndCommit to exclude the committed messages by commitIdentifier.
  1. import org.apache.paimon.data.BinaryString;
  2. import org.apache.paimon.data.GenericRow;
  3. import org.apache.paimon.table.Table;
  4. import org.apache.paimon.table.sink.CommitMessage;
  5. import org.apache.paimon.table.sink.StreamTableCommit;
  6. import org.apache.paimon.table.sink.StreamTableWrite;
  7. import org.apache.paimon.table.sink.StreamWriteBuilder;
  9. import java.util.List;
  11. public class StreamWriteTable {
  13.     public static void main(String[] args) throws Exception {
  14.         // 1. Create a WriteBuilder (Serializable)
  15.         Table table = GetTable.getTable();
  16.         StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder();
  18.         // 2. Write records in distributed tasks
  19.         StreamTableWrite write = writeBuilder.newWrite();
  20.         // commitIdentifier like Flink checkpointId
  21.         long commitIdentifier = 0;
  23.         while (true) {
  24.             GenericRow record1 = GenericRow.of(BinaryString.fromString("Alice"), 12);
  25.             GenericRow record2 = GenericRow.of(BinaryString.fromString("Bob"), 5);
  26.             GenericRow record3 = GenericRow.of(BinaryString.fromString("Emily"), 18);
  28.             // If this is a distributed write, you can use writeBuilder.newWriteSelector.
  29.             // WriteSelector determines to which logical downstream writers a record should be written to.
  30.             // If it returns empty, no data distribution is required.
  32.             write.write(record1);
  33.             write.write(record2);
  34.             write.write(record3);
  35.             List<CommitMessage> messages = write.prepareCommit(false, commitIdentifier);
  36.             commitIdentifier++;
  38.             // 3. Collect all CommitMessages to a global node and commit
  39.             StreamTableCommit commit = writeBuilder.newCommit();
  40.             commit.commit(commitIdentifier, messages);
  42.             // 4. When failure occurs and you're not sure if the commit process is successful,
  43.             //    you can use `filterAndCommit` to retry the commit process.
  44.             //    Succeeded commits will be automatically skipped.
  45.             /*
  46.             Map<Long, List<CommitMessage>> commitIdentifiersAndMessages = new HashMap<>();
  47.             commitIdentifiersAndMessages.put(commitIdentifier, messages);
  48.             commit.filterAndCommit(commitIdentifiersAndMessages);
  49.             */
  51.             Thread.sleep(1000);
  52.         }
  53.     }
  54. }

Data Types

JavaPaimon
booleanboolean
bytebyte
shortshort
intint
longlong
floatfloat
doubledouble
stringorg.apache.paimon.data.BinaryString
decimalorg.apache.paimon.data.Decimal
timestamporg.apache.paimon.data.Timestamp
byte[]byte[]
arrayorg.apache.paimon.data.InternalArray
maporg.apache.paimon.data.InternalMap
InternalRoworg.apache.paimon.data.InternalRow

Predicate Types

SQL PredicatePaimon Predicate
andorg.apache.paimon.predicate.PredicateBuilder.And
ororg.apache.paimon.predicate.PredicateBuilder.Or
is nullorg.apache.paimon.predicate.PredicateBuilder.IsNull
is not nullorg.apache.paimon.predicate.PredicateBuilder.IsNotNull
inorg.apache.paimon.predicate.PredicateBuilder.In
not inorg.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

Create Database

You can use the catalog to create databases. The created databases are persistence in the file system.

  1. import org.apache.paimon.catalog.Catalog;
  3. public class CreateDatabase {
  5.     public static void main(String[] args) {
  6.         try {
  7.             Catalog catalog = CreateCatalog.createFilesystemCatalog();
  8.             catalog.createDatabase("my_db", false);
  9.         } catch (Catalog.DatabaseAlreadyExistException e) {
  10.             // do something
  11.         }
  12.     }
  13. }

Determine Whether Database Exists

You can use the catalog to determine whether the database exists

  1. import org.apache.paimon.catalog.Catalog;
  3. public class DatabaseExists {
  5.     public static void main(String[] args) {
  6.         Catalog catalog = CreateCatalog.createFilesystemCatalog();
  7.         boolean exists = catalog.databaseExists("my_db");
  8.     }
  9. }

List Databases

You can use the catalog to list databases.

  1. import org.apache.paimon.catalog.Catalog;
  3. import java.util.List;
  5. public class ListDatabases {
  7.     public static void main(String[] args) {
  8.         Catalog catalog = CreateCatalog.createFilesystemCatalog();
  9.         List<String> databases = catalog.listDatabases();
  10.     }
  11. }

Drop Database

You can use the catalog to drop databases.

  1. import org.apache.paimon.catalog.Catalog;
  3. public class DropDatabase {
  5.     public static void main(String[] args) {
  6.         try {
  7.             Catalog catalog = CreateCatalog.createFilesystemCatalog();
  8.             catalog.dropDatabase("my_db", false, true);
  9.         } catch (Catalog.DatabaseNotEmptyException e) {
  10.             // do something
  11.         } catch (Catalog.DatabaseNotExistException e) {
  12.             // do something
  13.         }
  14.     }
  15. }

Determine Whether Table Exists

You can use the catalog to determine whether the table exists

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.Identifier;
  4. public class TableExists {
  6.     public static void main(String[] args) {
  7.         Identifier identifier = Identifier.create("my_db", "my_table");
  8.         Catalog catalog = CreateCatalog.createFilesystemCatalog();
  9.         boolean exists = catalog.tableExists(identifier);
  10.     }
  11. }

List Tables

You can use the catalog to list tables.

  1. import org.apache.paimon.catalog.Catalog;
  3. import java.util.List;
  5. public class ListTables {
  7.     public static void main(String[] args) {
  8.         try {
  9.             Catalog catalog = CreateCatalog.createFilesystemCatalog();
  10.             List<String> tables = catalog.listTables("my_db");
  11.         } catch (Catalog.DatabaseNotExistException e) {
  12.             // do something
  13.         }
  14.     }
  15. }

Drop Table

You can use the catalog to drop table.

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.Identifier;
  4. public class DropTable {
  6.     public static void main(String[] args) {
  7.         Identifier identifier = Identifier.create("my_db", "my_table");
  8.         try {
  9.             Catalog catalog = CreateCatalog.createFilesystemCatalog();
  10.             catalog.dropTable(identifier, false);
  11.         } catch (Catalog.TableNotExistException e) {
  12.             // do something
  13.         }
  14.     }
  15. }

Rename Table

You can use the catalog to rename a table.

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.Identifier;
  4. public class RenameTable {
  6.     public static void main(String[] args) {
  7.         Identifier fromTableIdentifier = Identifier.create("my_db", "my_table");
  8.         Identifier toTableIdentifier = Identifier.create("my_db", "test_table");
  9.         try {
  10.             Catalog catalog = CreateCatalog.createFilesystemCatalog();
  11.             catalog.renameTable(fromTableIdentifier, toTableIdentifier, false);
  12.         } catch (Catalog.TableAlreadyExistException e) {
  13.             // do something
  14.         } catch (Catalog.TableNotExistException e) {
  15.             // do something
  16.         }
  17.     }
  18. }

Alter Table

You can use the catalog to alter a table, but you need to pay attention to the following points.

  • 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.
  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.Identifier;
  3. import org.apache.paimon.schema.Schema;
  4. import org.apache.paimon.schema.SchemaChange;
  5. import org.apache.paimon.types.DataField;
  6. import org.apache.paimon.types.DataTypes;
  8. import com.google.common.collect.Lists;
  10. import java.util.Arrays;
  11. import java.util.HashMap;
  12. import java.util.Map;
  14. public class AlterTable {
  16.     public static void main(String[] args) {
  17.         Identifier identifier = Identifier.create("my_db", "my_table");
  19.         Map<String, String> options = new HashMap<>();
  20.         options.put("bucket", "4");
  21.         options.put("compaction.max.file-num", "40");
  23.         Catalog catalog = CreateCatalog.createFilesystemCatalog();
  24.         catalog.createDatabase("my_db", false);
  26.         try {
  27.             catalog.createTable(
  28.                 identifier,
  29.                 new Schema(
  30.                     Lists.newArrayList(
  31.                         new DataField(0, "col1", DataTypes.STRING(), "field1"),
  32.                         new DataField(1, "col2", DataTypes.STRING(), "field2"),
  33.                         new DataField(2, "col3", DataTypes.STRING(), "field3"),
  34.                         new DataField(3, "col4", DataTypes.BIGINT(), "field4"),
  35.                         new DataField(
  36.                             4,
  37.                             "col5",
  38.                             DataTypes.ROW(
  39.                                 new DataField(
  40.                                     5, "f1", DataTypes.STRING(), "f1"),
  41.                                 new DataField(
  42.                                     6, "f2", DataTypes.STRING(), "f2"),
  43.                                 new DataField(
  44.                                     7, "f3", DataTypes.STRING(), "f3")),
  45.                             "field5"),
  46.                         new DataField(8, "col6", DataTypes.STRING(), "field6")),
  47.                     Lists.newArrayList("col1"), // partition keys
  48.                     Lists.newArrayList("col1", "col2"), // primary key
  49.                     options,
  50.                     "table comment"),
  51.                 false);
  52.         } catch (Catalog.TableAlreadyExistException e) {
  53.             // do something
  54.         } catch (Catalog.DatabaseNotExistException e) {
  55.             // do something
  56.         }
  58.         // add option
  59.         SchemaChange addOption = SchemaChange.setOption("snapshot.time-retained", "2h");
  60.         // remove option
  61.         SchemaChange removeOption = SchemaChange.removeOption("compaction.max.file-num");
  62.         // add column
  63.         SchemaChange addColumn = SchemaChange.addColumn("col1_after", DataTypes.STRING());
  64.         // add a column after col1
  65.         SchemaChange.Move after = SchemaChange.Move.after("col1_after", "col1");
  66.         SchemaChange addColumnAfterField =
  67.             SchemaChange.addColumn("col7", DataTypes.STRING(), "", after);
  68.         // rename column
  69.         SchemaChange renameColumn = SchemaChange.renameColumn("col3", "col3_new_name");
  70.         // drop column
  71.         SchemaChange dropColumn = SchemaChange.dropColumn("col6");
  72.         // update column comment
  73.         SchemaChange updateColumnComment =
  74.             SchemaChange.updateColumnComment(new String[] {"col4"}, "col4 field");
  75.         // update nested column comment
  76.         SchemaChange updateNestedColumnComment =
  77.             SchemaChange.updateColumnComment(new String[] {"col5", "f1"}, "col5 f1 field");
  78.         // update column type
  79.         SchemaChange updateColumnType = SchemaChange.updateColumnType("col4", DataTypes.DOUBLE());
  80.         // update column position, you need to pass in a parameter of type Move
  81.         SchemaChange updateColumnPosition =
  82.             SchemaChange.updateColumnPosition(SchemaChange.Move.first("col4"));
  83.         // update column nullability
  84.         SchemaChange updateColumnNullability =
  85.             SchemaChange.updateColumnNullability(new String[] {"col4"}, false);
  86.         // update nested column nullability
  87.         SchemaChange updateNestedColumnNullability =
  88.             SchemaChange.updateColumnNullability(new String[] {"col5", "f2"}, false);
  90.         SchemaChange[] schemaChanges =
  91.             new SchemaChange[] {
  92.                 addOption,
  93.                 removeOption,
  94.                 addColumn,
  95.                 addColumnAfterField,
  96.                 renameColumn,
  97.                 dropColumn,
  98.                 updateColumnComment,
  99.                 updateNestedColumnComment,
  100.                 updateColumnType,
  101.                 updateColumnPosition,
  102.                 updateColumnNullability,
  103.                 updateNestedColumnNullability
  104.             };
  105.         try {
  106.             catalog.alterTable(identifier, Arrays.asList(schemaChanges), false);
  107.         } catch (Catalog.TableNotExistException e) {
  108.             // do something
  109.         } catch (Catalog.ColumnAlreadyExistException e) {
  110.             // do something
  111.         } catch (Catalog.ColumnNotExistException e) {
  112.             // do something
  113.         }
  114.     }
  115. }

Table metadata:

  • name return a name string to identify this table.
  • rowType return the current row type of this table containing a sequence of table’s fields.
  • partitionKeys returns the partition keys of this table.
  • parimaryKeys returns the primary keys of this table.
  • options returns the configuration of this table in a map of key-value.
  • comment returns the optional comment of this table.
  • copy return a new table by applying dynamic options to this table.