This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ca5d7d18fe [python] Add branch CLI commands (#8325)
ca5d7d18fe is described below
commit ca5d7d18fef8ef2e583e5e8218691d0ea5c2167d
Author: chaoyang <[email protected]>
AuthorDate: Tue Jun 23 11:50:01 2026 +0800
[python] Add branch CLI commands (#8325)
---
docs/docs/pypaimon/cli.md | 54 +++++
paimon-python/pypaimon/cli/cli.py | 4 +
paimon-python/pypaimon/cli/cli_branch.py | 228 +++++++++++++++++++++
paimon-python/pypaimon/tests/cli_branch_test.py | 182 ++++++++++++++++
.../tests/filesystem_catalog_branch_test.py | 34 +--
5 files changed, 487 insertions(+), 15 deletions(-)
diff --git a/docs/docs/pypaimon/cli.md b/docs/docs/pypaimon/cli.md
index 94c0fdd688..df5b7e0fee 100644
--- a/docs/docs/pypaimon/cli.md
+++ b/docs/docs/pypaimon/cli.md
@@ -785,3 +785,57 @@ SQL statements end with `;` and can span multiple lines.
The continuation prompt
| `exit` / `quit` | Exit the REPL |
For more details on SQL syntax and the Python API, see [SQL Query](./sql).
+
+## Branch Commands
+
+Manage branches on a table. Branches are independent lines of a table that can
be created from the current state or from a tag, and later fast-forwarded back
into main.
+
+```shell
+paimon branch <create|list|delete|rename|fast-forward> mydb.users ...
+```
+
+### Branch Create
+
+```shell
+# Create a branch from the current state
+paimon branch create mydb.users b1
+
+# Create a branch from an existing tag
+paimon branch create mydb.users b1 --tag v1
+```
+
+Options:
+- `--tag, -t`: Create the branch from this tag (default: current state)
+
+### Branch List
+
+```shell
+# List all branches
+paimon branch list mydb.users
+
+# JSON output
+paimon branch list mydb.users --format json
+```
+
+Options:
+- `--format, -f`: Output format, `table` (default) or `json`
+
+### Branch Delete
+
+```shell
+paimon branch delete mydb.users b1
+```
+
+### Branch Rename
+
+```shell
+paimon branch rename mydb.users b1 b2
+```
+
+### Branch Fast-Forward
+
+Fast-forward the main branch to the given branch (main adopts the branch's
snapshots).
+
+```shell
+paimon branch fast-forward mydb.users b1
+```
diff --git a/paimon-python/pypaimon/cli/cli.py
b/paimon-python/pypaimon/cli/cli.py
index 3e1547fe39..66a1d37b9d 100644
--- a/paimon-python/pypaimon/cli/cli.py
+++ b/paimon-python/pypaimon/cli/cli.py
@@ -125,6 +125,10 @@ def main():
from pypaimon.cli.cli_sql import add_sql_subcommand
add_sql_subcommand(subparsers)
+ # Branch commands
+ from pypaimon.cli.cli_branch import add_branch_subcommands
+ add_branch_subcommands(subparsers)
+
args = parser.parse_args()
if args.command is None:
diff --git a/paimon-python/pypaimon/cli/cli_branch.py
b/paimon-python/pypaimon/cli/cli_branch.py
new file mode 100644
index 0000000000..d6b769a77d
--- /dev/null
+++ b/paimon-python/pypaimon/cli/cli_branch.py
@@ -0,0 +1,228 @@
+# 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.
+
+"""Branch commands for the Paimon CLI.
+
+Adds the top-level ``branch {create,list,delete,rename,fast-forward}``
+command. All operations go through the Catalog layer so they get typed
+exceptions and work for both filesystem and REST catalogs. Command and
+argument names mirror the Java branch procedures (``create_branch`` /
+``delete_branch`` / ``rename_branch`` / ``fast_forward``).
+"""
+
+import json
+import sys
+
+from pypaimon.catalog.catalog_exception import (BranchAlreadyExistException,
+ BranchNotExistException,
+ TableNotExistException,
+ TagNotExistException)
+
+
+def _open_catalog(args):
+ """Load config, build the catalog and validate the ``database.table`` id.
+
+ Returns ``(catalog, identifier)``. On any failure, prints to stderr and
+ exits with a non-zero status (matching the other CLI commands).
+ """
+ from pypaimon.cli.cli import create_catalog, load_catalog_config
+
+ identifier = args.table
+ if len(identifier.split('.')) != 2:
+ print("Error: Invalid table identifier '{}'. Expected format: "
+ "'database.table'".format(identifier), file=sys.stderr)
+ sys.exit(1)
+ try:
+ catalog = create_catalog(load_catalog_config(args.config))
+ except Exception as e:
+ print("Error: {}".format(e), file=sys.stderr)
+ sys.exit(1)
+ return catalog, identifier
+
+
+def cmd_branch_create(args):
+ """Execute ``branch create``."""
+ catalog, identifier = _open_catalog(args)
+ try:
+ catalog.create_branch(identifier, args.branch_name, tag_name=args.tag)
+ except TableNotExistException:
+ print("Error: Table '{}' does not exist.".format(identifier),
+ file=sys.stderr)
+ sys.exit(1)
+ except BranchAlreadyExistException:
+ print("Error: Branch '{}' already exists.".format(args.branch_name),
+ file=sys.stderr)
+ sys.exit(1)
+ except TagNotExistException:
+ print("Error: Tag '{}' does not exist.".format(args.tag),
+ file=sys.stderr)
+ sys.exit(1)
+ except Exception as e:
+ print("Error: Failed to create branch: {}".format(e), file=sys.stderr)
+ sys.exit(1)
+ if args.tag is not None:
+ print("Branch '{}' created from tag '{}' on table '{}'.".format(
+ args.branch_name, args.tag, identifier))
+ else:
+ print("Branch '{}' created on table '{}'.".format(
+ args.branch_name, identifier))
+
+
+def cmd_branch_delete(args):
+ """Execute ``branch delete``."""
+ catalog, identifier = _open_catalog(args)
+ try:
+ catalog.drop_branch(identifier, args.branch_name)
+ except TableNotExistException:
+ print("Error: Table '{}' does not exist.".format(identifier),
+ file=sys.stderr)
+ sys.exit(1)
+ except BranchNotExistException:
+ print("Error: Branch '{}' does not exist.".format(args.branch_name),
+ file=sys.stderr)
+ sys.exit(1)
+ except Exception as e:
+ print("Error: Failed to delete branch: {}".format(e), file=sys.stderr)
+ sys.exit(1)
+ print("Branch '{}' deleted from table '{}'.".format(
+ args.branch_name, identifier))
+
+
+def cmd_branch_list(args):
+ """Execute ``branch list``."""
+ catalog, identifier = _open_catalog(args)
+ try:
+ branches = catalog.list_branches(identifier)
+ except TableNotExistException:
+ print("Error: Table '{}' does not exist.".format(identifier),
+ file=sys.stderr)
+ sys.exit(1)
+ except Exception as e:
+ print("Error: Failed to list branches: {}".format(e), file=sys.stderr)
+ sys.exit(1)
+
+ if args.format == 'json':
+ print(json.dumps(branches, ensure_ascii=False))
+ elif not branches:
+ print("No branches found.")
+ else:
+ for branch in branches:
+ print(branch)
+
+
+def cmd_branch_rename(args):
+ """Execute ``branch rename``."""
+ catalog, identifier = _open_catalog(args)
+ try:
+ catalog.rename_branch(identifier, args.from_branch, args.to_branch)
+ except TableNotExistException:
+ print("Error: Table '{}' does not exist.".format(identifier),
+ file=sys.stderr)
+ sys.exit(1)
+ except BranchNotExistException:
+ print("Error: Branch '{}' does not exist.".format(args.from_branch),
+ file=sys.stderr)
+ sys.exit(1)
+ except BranchAlreadyExistException:
+ print("Error: Branch '{}' already exists.".format(args.to_branch),
+ file=sys.stderr)
+ sys.exit(1)
+ except Exception as e:
+ print("Error: Failed to rename branch: {}".format(e), file=sys.stderr)
+ sys.exit(1)
+ print("Branch '{}' renamed to '{}' on table '{}'.".format(
+ args.from_branch, args.to_branch, identifier))
+
+
+def cmd_branch_fast_forward(args):
+ """Execute ``branch fast-forward``."""
+ catalog, identifier = _open_catalog(args)
+ try:
+ catalog.fast_forward(identifier, args.branch_name)
+ except TableNotExistException:
+ print("Error: Table '{}' does not exist.".format(identifier),
+ file=sys.stderr)
+ sys.exit(1)
+ except BranchNotExistException:
+ print("Error: Branch '{}' does not exist.".format(args.branch_name),
+ file=sys.stderr)
+ sys.exit(1)
+ except Exception as e:
+ print("Error: Failed to fast-forward: {}".format(e), file=sys.stderr)
+ sys.exit(1)
+ print("Fast-forwarded table '{}' to branch '{}'.".format(
+ identifier, args.branch_name))
+
+
+def add_branch_subcommands(subparsers):
+ """Register the top-level ``branch <command>`` command.
+
+ Args:
+ subparsers: The subparsers object from the main argument parser.
+ """
+ branch_parser = subparsers.add_parser('branch', help='Branch operations')
+ branch_subparsers = branch_parser.add_subparsers(
+ dest='branch_command', help='Branch commands')
+
+ # branch create
+ create_parser = branch_subparsers.add_parser(
+ 'create', help='Create a branch on a table')
+ create_parser.add_argument(
+ 'table', help='Table identifier in format: database.table')
+ create_parser.add_argument(
+ 'branch_name', help='Name of the branch to create')
+ create_parser.add_argument(
+ '--tag', '-t', default=None,
+ help='Create the branch from this tag (default: current state)')
+ create_parser.set_defaults(func=cmd_branch_create)
+
+ # branch list
+ list_parser = branch_subparsers.add_parser(
+ 'list', help='List branches of a table')
+ list_parser.add_argument(
+ 'table', help='Table identifier in format: database.table')
+ list_parser.add_argument(
+ '--format', '-f', choices=['table', 'json'], default='table',
+ help='Output format: table (default) or json')
+ list_parser.set_defaults(func=cmd_branch_list)
+
+ # branch delete
+ delete_parser = branch_subparsers.add_parser(
+ 'delete', help='Delete a branch from a table')
+ delete_parser.add_argument(
+ 'table', help='Table identifier in format: database.table')
+ delete_parser.add_argument(
+ 'branch_name', help='Name of the branch to delete')
+ delete_parser.set_defaults(func=cmd_branch_delete)
+
+ # branch rename
+ rename_parser = branch_subparsers.add_parser(
+ 'rename', help='Rename a branch')
+ rename_parser.add_argument(
+ 'table', help='Table identifier in format: database.table')
+ rename_parser.add_argument('from_branch', help='Current branch name')
+ rename_parser.add_argument('to_branch', help='New branch name')
+ rename_parser.set_defaults(func=cmd_branch_rename)
+
+ # branch fast-forward
+ ff_parser = branch_subparsers.add_parser(
+ 'fast-forward', help='Fast-forward main to a branch')
+ ff_parser.add_argument(
+ 'table', help='Table identifier in format: database.table')
+ ff_parser.add_argument(
+ 'branch_name', help='Name of the branch to fast-forward to')
+ ff_parser.set_defaults(func=cmd_branch_fast_forward)
diff --git a/paimon-python/pypaimon/tests/cli_branch_test.py
b/paimon-python/pypaimon/tests/cli_branch_test.py
new file mode 100644
index 0000000000..c21e44779d
--- /dev/null
+++ b/paimon-python/pypaimon/tests/cli_branch_test.py
@@ -0,0 +1,182 @@
+# 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.
+
+"""Integration tests for the top-level ``branch`` CLI command."""
+
+import json
+import os
+import shutil
+import tempfile
+import unittest
+from io import StringIO
+from unittest.mock import patch
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.cli.cli import main
+
+
+class CliBranchTest(unittest.TestCase):
+
+ def setUp(self):
+ self.tempdir = tempfile.mkdtemp(prefix="cli_branch_")
+ self.warehouse = os.path.join(self.tempdir, 'warehouse')
+ self.catalog = CatalogFactory.create({'warehouse': self.warehouse})
+ self.catalog.create_database('db', True)
+
+ pa_schema = pa.schema([('id', pa.int32()), ('name', pa.string())])
+ self.catalog.create_table(
+ 'db.t', Schema.from_pyarrow_schema(pa_schema), False)
+ # One commit so there is a snapshot, plus a tag for from-tag tests.
+ table = self.catalog.get_table('db.t')
+ wb = table.new_batch_write_builder()
+ w = wb.new_write()
+ c = wb.new_commit()
+ w.write_arrow(pa.Table.from_pylist([{'id': 1, 'name': 'a'}],
+ schema=pa_schema))
+ c.commit(w.prepare_commit())
+ w.close()
+ c.close()
+ table.create_tag("v1")
+
+ self.config_file = os.path.join(self.tempdir, 'paimon.yaml')
+ with open(self.config_file, 'w') as f:
+ f.write("metastore: filesystem\nwarehouse: {}\n".format(
+ self.warehouse))
+
+ def tearDown(self):
+ shutil.rmtree(self.tempdir, ignore_errors=True)
+
+ def _run(self, *argv):
+ """Run the CLI; return (stdout, stderr, exit_code)."""
+ out, err = StringIO(), StringIO()
+ code = 0
+ full = ['paimon', '-c', self.config_file] + list(argv)
+ with patch('sys.argv', full):
+ with patch('sys.stdout', out), patch('sys.stderr', err):
+ try:
+ main()
+ except SystemExit as e:
+ code = 0 if e.code is None else (
+ e.code if isinstance(e.code, int) else 1)
+ return out.getvalue(), err.getvalue(), code
+
+ # -- create + list -------------------------------------------------------
+
+ def test_create_then_list(self):
+ out, _, code = self._run('branch', 'create', 'db.t', 'b1')
+ self.assertEqual(0, code)
+ self.assertIn("created", out)
+
+ out, _, code = self._run('branch', 'list', 'db.t')
+ self.assertEqual(0, code)
+ self.assertIn("b1", out)
+
+ def test_create_from_tag(self):
+ out, _, code = self._run(
+ 'branch', 'create', 'db.t', 'b1', '--tag', 'v1')
+ self.assertEqual(0, code)
+ self.assertIn("from tag 'v1'", out)
+ out, _, _ = self._run('branch', 'list', 'db.t')
+ self.assertIn("b1", out)
+
+ def test_create_from_missing_tag(self):
+ _, err, code = self._run(
+ 'branch', 'create', 'db.t', 'b1', '--tag', 'absent')
+ self.assertEqual(1, code)
+ self.assertIn("does not exist", err)
+
+ def test_create_duplicate_raises(self):
+ self._run('branch', 'create', 'db.t', 'b1')
+ _, err, code = self._run('branch', 'create', 'db.t', 'b1')
+ self.assertEqual(1, code)
+ self.assertIn("already exists", err)
+
+ def test_list_empty(self):
+ out, _, code = self._run('branch', 'list', 'db.t')
+ self.assertEqual(0, code)
+ self.assertIn("No branches found.", out)
+
+ def test_list_json(self):
+ self._run('branch', 'create', 'db.t', 'b1')
+ self._run('branch', 'create', 'db.t', 'b2')
+ out, _, code = self._run('branch', 'list', 'db.t', '--format', 'json')
+ self.assertEqual(0, code)
+ self.assertEqual({"b1", "b2"}, set(json.loads(out)))
+
+ # -- delete --------------------------------------------------------------
+
+ def test_delete(self):
+ self._run('branch', 'create', 'db.t', 'b1')
+ out, _, code = self._run('branch', 'delete', 'db.t', 'b1')
+ self.assertEqual(0, code)
+ self.assertIn("deleted", out)
+ out, _, _ = self._run('branch', 'list', 'db.t')
+ self.assertNotIn("b1", out)
+
+ def test_delete_not_exists(self):
+ _, err, code = self._run('branch', 'delete', 'db.t', 'absent')
+ self.assertEqual(1, code)
+ self.assertIn("does not exist", err)
+
+ # -- rename --------------------------------------------------------------
+
+ def test_rename(self):
+ self._run('branch', 'create', 'db.t', 'b1')
+ out, _, code = self._run('branch', 'rename', 'db.t', 'b1', 'b2')
+ self.assertEqual(0, code)
+ self.assertIn("renamed", out)
+ out, _, _ = self._run('branch', 'list', 'db.t')
+ self.assertIn("b2", out)
+ self.assertNotIn("b1", out)
+
+ def test_rename_from_missing(self):
+ _, err, code = self._run('branch', 'rename', 'db.t', 'absent', 'b2')
+ self.assertEqual(1, code)
+ self.assertIn("does not exist", err)
+
+ def test_rename_to_existing(self):
+ self._run('branch', 'create', 'db.t', 'b1')
+ self._run('branch', 'create', 'db.t', 'b2')
+ _, err, code = self._run('branch', 'rename', 'db.t', 'b1', 'b2')
+ self.assertEqual(1, code)
+ self.assertIn("already exists", err)
+
+ # -- fast-forward --------------------------------------------------------
+
+ def test_fast_forward(self):
+ self._run('branch', 'create', 'db.t', 'b1', '--tag', 'v1')
+ out, _, code = self._run('branch', 'fast-forward', 'db.t', 'b1')
+ self.assertEqual(0, code)
+ self.assertIn("Fast-forwarded", out)
+
+ def test_fast_forward_missing(self):
+ _, err, code = self._run('branch', 'fast-forward', 'db.t', 'absent')
+ self.assertEqual(1, code)
+ self.assertIn("does not exist", err)
+
+ # -- bad input -----------------------------------------------------------
+
+ def test_invalid_identifier(self):
+ _, err, code = self._run('branch', 'create', 'nodot', 'b1')
+ self.assertEqual(1, code)
+ self.assertIn("Invalid table identifier", err)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
b/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
index b60fe97a47..2129f811fa 100644
--- a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
+++ b/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
@@ -94,15 +94,18 @@ class FileSystemCatalogBranchCRUDTest(unittest.TestCase):
self.identifier, "b1", tag_name="absent_tag")
self.assertEqual(cm.exception.tag, "absent_tag")
- # NOTE: ``test_create_branch_from_existing_tag`` (a true happy-path
- # ``create_branch(tag_name=...)``) is not included here. The
- # ``FileSystemBranchManager`` "from-tag" path has a pre-existing bug
- # (``branch_snapshot_manager`` is constructed without switching to
- # the new branch's path, so ``copy_file(src, dst)`` ends up with
- # ``src == dst`` and raises ``SameFileError``). That's a manager-
- # level fix, not in the scope of this catalog-layer thin wrapper.
- # Catalog-layer error translation for the from-tag path is still
- # covered by ``test_create_branch_from_nonexistent_tag_raises``.
+ def test_create_branch_from_existing_tag(self):
+ # The from-tag happy path: create_tag then create_branch(tag_name=...)
+ # must land the branch files under ``branch/branch-<name>/`` and not
+ # raise (regresses the historical src == dst SameFileError).
+ table = self.catalog.get_table(self.identifier)
+ table.create_tag("t1")
+ self.catalog.create_branch(self.identifier, "b1", tag_name="t1")
+ self.assertIn("b1", self.catalog.list_branches(self.identifier))
+ branch_root = "{}/branch/branch-b1".format(
+ table.table_path.rstrip('/'))
+ self.assertTrue(os.path.isdir(branch_root))
+ self.assertTrue(os.path.isfile("{}/tag/tag-t1".format(branch_root)))
# -- list -----------------------------------------------------------------
@@ -164,12 +167,13 @@ class FileSystemCatalogBranchCRUDTest(unittest.TestCase):
self.catalog.fast_forward(self.identifier, "absent")
self.assertEqual(cm.exception.branch, "absent")
- # NOTE: a true happy-path ``fast_forward`` end-to-end test is not
- # included here for the same reason as the create-branch-from-tag
- # case above — it requires the manager-level fix to the from-tag
- # path (so the branch carries a snapshot for fast-forward to move).
- # Catalog-layer error translation is covered by the missing-branch
- # case above.
+ def test_fast_forward_after_create_branch_from_tag(self):
+ # Happy path: create a branch from a tag, then fast-forward main to
+ # it. Must not raise (regresses the historical src == dst error).
+ table = self.catalog.get_table(self.identifier)
+ table.create_tag("t1")
+ self.catalog.create_branch(self.identifier, "b1", tag_name="t1")
+ self.catalog.fast_forward(self.identifier, "b1")
if __name__ == "__main__":
