Motivation
In data streaming process there may be data errors and other issues, and we need to correct the data in the flow. This situation is very common and important. However, in this process, we do not want to affect existing data processing to avoid impact on users. We need to create a new data streaming process and wait for it to catch up with the data and replace the original data streaming process. The main operations can be divided into the following steps:
Create a replica table based on the specified tag/snapshot of upstream and downstream Paimon Tables
Resubmit all streaming jobs, incremental or full recovery starting from the specified offset
I think we need to support branching in Paimon. Then we could create replica tables to avoid copying all data from specified tables and increase storage space.
Besides the above, branching in Paimon can also be used to enhance tag. for Tag simulation of traditional Hive partition tables, provide data correction capabilities on the basis of Tag, which can be used to supplement data and achieve precise segmentation capabilities.
Above all, the branch we would like to introduce in Paimon has the following abilities:
Each table only has one main branch, and other branches can only be created from the specified tag of the main branch
Create or delete a branch for tables in Paimon, and create a tag for a specified branch.
Update schema for the branch, such as altering tables to add/drop columns.
Jobs can streaming/batch read from and write data to the branch
There are merge and replace operations from branch to main, and after replace main with given branch, the previous main branch will be deleted.
Architecture
Data storage structure in Paimon is divided into five components: schema, data file, manifest file and list, snapshot, tag.
eyJleHRTcnZJbnRlZ1R5cGUiOiIiLCJnQ2xpZW50SWQiOiIiLCJjcmVhdG9yTmFtZSI6IkZhbmcgWW9uZyIsIm91dHB1dFR5cGUiOiJibG9jayIsImxhc3RNb2RpZmllck5hbWUiOiJGYW5nIFlvbmciLCJsYW5ndWFnZSI6ImVuIiwiZGlhZ3JhbURpc3BsYXlOYW1lIjoiIiwic0ZpbGVJZCI6IiIsImF0dElkIjoiMjYzNDI4NjQ4IiwiZGlhZ3JhbU5hbWUiOiIxIiwiYXNwZWN0IjoiIiwibGlua3MiOiJhdXRvIiwiY2VvTmFtZSI6IlBJUC05OiBTdXBwb3J0IEJyYW5jaCIsInRic3R5bGUiOiJ0b3AiLCJjYW5Db21tZW50IjpmYWxzZSwiZGlhZ3JhbVVybCI6IiIsImNzdkZpbGVVcmwiOiIiLCJib3JkZXIiOnRydWUsIm1heFNjYWxlIjoiMSIsIm93bmluZ1BhZ2VJZCI6MjYzNDI3MzgzLCJlZGl0YWJsZSI6ZmFsc2UsImNlb0lkIjoyNjM0Mjg2NjIsInBhZ2VJZCI6IiIsImxib3giOnRydWUsInNlcnZlckNvbmZpZyI6eyJlbWFpbHByZXZpZXciOiIxIn0sIm9kcml2ZUlkIjoiIiwicmV2aXNpb24iOjEsIm1hY3JvSWQiOiJmZTUyMjQ4Mi00N2YzLTQ4MGEtYmU2NS00NjdhMWU0NjM0YzQiLCJwcmV2aWV3TmFtZSI6IjEucG5nIiwibGljZW5zZVN0YXR1cyI6Ik9LIiwic2VydmljZSI6IiIsImlzVGVtcGxhdGUiOiIiLCJ3aWR0aCI6IjYyMSIsInNpbXBsZVZpZXdlciI6ZmFsc2UsImxhc3RNb2RpZmllZCI6MTY5NDA4NzYyMDAwMCwiZXhjZWVkUGFnZVdpZHRoIjpmYWxzZSwib0NsaWVudElkIjoiIn0=
To support the above branch capabilities, we would like to introduce branches for snapshot, tag and schema as follows.
eyJleHRTcnZJbnRlZ1R5cGUiOiIiLCJnQ2xpZW50SWQiOiIiLCJjcmVhdG9yTmFtZSI6IkZhbmcgWW9uZyIsIm91dHB1dFR5cGUiOiJibG9jayIsImxhc3RNb2RpZmllck5hbWUiOiJGYW5nIFlvbmciLCJsYW5ndWFnZSI6ImVuIiwiZGlhZ3JhbURpc3BsYXlOYW1lIjoiIiwic0ZpbGVJZCI6IiIsImF0dElkIjoiMjYzNDI4NjUwIiwiZGlhZ3JhbU5hbWUiOiIyIiwiYXNwZWN0IjoiIiwibGlua3MiOiJhdXRvIiwiY2VvTmFtZSI6IlBJUC05OiBTdXBwb3J0IEJyYW5jaCIsInRic3R5bGUiOiJ0b3AiLCJjYW5Db21tZW50IjpmYWxzZSwiZGlhZ3JhbVVybCI6IiIsImNzdkZpbGVVcmwiOiIiLCJib3JkZXIiOnRydWUsIm1heFNjYWxlIjoiMSIsIm93bmluZ1BhZ2VJZCI6MjYzNDI3MzgzLCJlZGl0YWJsZSI6ZmFsc2UsImNlb0lkIjoyNjM0Mjg2NjIsInBhZ2VJZCI6IiIsImxib3giOnRydWUsInNlcnZlckNvbmZpZyI6eyJlbWFpbHByZXZpZXciOiIxIn0sIm9kcml2ZUlkIjoiIiwicmV2aXNpb24iOjEsIm1hY3JvSWQiOiI4ZTM3YTAxOC1mNzlkLTRkMTctODRjYy03MDZjMzYxODZhMzAiLCJwcmV2aWV3TmFtZSI6IjIucG5nIiwibGljZW5zZVN0YXR1cyI6Ik9LIiwic2VydmljZSI6IiIsImlzVGVtcGxhdGUiOiIiLCJ3aWR0aCI6IjU5MSIsInNpbXBsZVZpZXdlciI6ZmFsc2UsImxhc3RNb2RpZmllZCI6MTY5NTAwNDkwMzAwMCwiZXhjZWVkUGFnZVdpZHRoIjpmYWxzZSwib0NsaWVudElkIjoiIn0=
There is a main branch file in the branch directory and it has the main branch name in the file. Besides that, there will be multiple branch directories and each branch has snapshot, tag and schema in its directory.
Create Branch
There will be a series of snapshots, tags and schemas in the main branch of a Paimon table. We can create a new branch with branch-name from the tag for the table. To do that, Paimon will create a new directory with the given branch name, copy the specified tag, snapshot and schema from the main branch to the new branch.
eyJleHRTcnZJbnRlZ1R5cGUiOiIiLCJnQ2xpZW50SWQiOiIiLCJjcmVhdG9yTmFtZSI6IkZhbmcgWW9uZyIsIm91dHB1dFR5cGUiOiJibG9jayIsImxhc3RNb2RpZmllck5hbWUiOiJGYW5nIFlvbmciLCJsYW5ndWFnZSI6ImVuIiwiZGlhZ3JhbURpc3BsYXlOYW1lIjoiIiwic0ZpbGVJZCI6IiIsImF0dElkIjoiMjYzNDI4NjUyIiwiZGlhZ3JhbU5hbWUiOiIzIiwiYXNwZWN0IjoiIiwibGlua3MiOiJhdXRvIiwiY2VvTmFtZSI6IlBJUC05OiBTdXBwb3J0IEJyYW5jaCIsInRic3R5bGUiOiJ0b3AiLCJjYW5Db21tZW50IjpmYWxzZSwiZGlhZ3JhbVVybCI6IiIsImNzdkZpbGVVcmwiOiIiLCJib3JkZXIiOnRydWUsIm1heFNjYWxlIjoiMSIsIm93bmluZ1BhZ2VJZCI6MjYzNDI3MzgzLCJlZGl0YWJsZSI6ZmFsc2UsImNlb0lkIjoyNjM0Mjg2NjIsInBhZ2VJZCI6IiIsImxib3giOnRydWUsInNlcnZlckNvbmZpZyI6eyJlbWFpbHByZXZpZXciOiIxIn0sIm9kcml2ZUlkIjoiIiwicmV2aXNpb24iOjEsIm1hY3JvSWQiOiIyZjkzNmNhNC1jYzA0LTQ2NjgtYmVmYS00ODdhZjNiZGYzNDMiLCJwcmV2aWV3TmFtZSI6IjMucG5nIiwibGljZW5zZVN0YXR1cyI6Ik9LIiwic2VydmljZSI6IiIsImlzVGVtcGxhdGUiOiIiLCJ3aWR0aCI6Ijg0MSIsInNpbXBsZVZpZXdlciI6ZmFsc2UsImxhc3RNb2RpZmllZCI6MTY5NDA4NzcwNDAwMCwiZXhjZWVkUGFnZVdpZHRoIjpmYWxzZSwib0NsaWVudElkIjoiIn0=
For example, when Branch-1 is created from tag-1, it should copy the relevant snapshot-4 and schema-1 for Branch-1. Branch-2 and Branch-3 will do the same thing for tag-7 and tag-11.
Operations In Branch
After a branch is created, streaming and batch jobs can read and write data in it. Like a regular table, we can also streaming and batch data from branch through time travel. After writing data to the branch, new snapshots and tags will be created. Users can also perform DDL for table branches, such as add/drop/alter columns. For example, we do these operations in Branch-1 to create new schemas, snapshots and tags.
eyJleHRTcnZJbnRlZ1R5cGUiOiIiLCJnQ2xpZW50SWQiOiIiLCJjcmVhdG9yTmFtZSI6IkZhbmcgWW9uZyIsIm91dHB1dFR5cGUiOiJibG9jayIsImxhc3RNb2RpZmllck5hbWUiOiJGYW5nIFlvbmciLCJsYW5ndWFnZSI6ImVuIiwiZGlhZ3JhbURpc3BsYXlOYW1lIjoiIiwic0ZpbGVJZCI6IiIsImF0dElkIjoiMjYzNDI4NjU0IiwiZGlhZ3JhbU5hbWUiOiI0IiwiYXNwZWN0IjoiIiwibGlua3MiOiJhdXRvIiwiY2VvTmFtZSI6IlBJUC05OiBTdXBwb3J0IEJyYW5jaCIsInRic3R5bGUiOiJ0b3AiLCJjYW5Db21tZW50IjpmYWxzZSwiZGlhZ3JhbVVybCI6IiIsImNzdkZpbGVVcmwiOiIiLCJib3JkZXIiOnRydWUsIm1heFNjYWxlIjoiMSIsIm93bmluZ1BhZ2VJZCI6MjYzNDI3MzgzLCJlZGl0YWJsZSI6ZmFsc2UsImNlb0lkIjoyNjM0Mjg2NjIsInBhZ2VJZCI6IiIsImxib3giOnRydWUsInNlcnZlckNvbmZpZyI6eyJlbWFpbHByZXZpZXciOiIxIn0sIm9kcml2ZUlkIjoiIiwicmV2aXNpb24iOjEsIm1hY3JvSWQiOiI3N2U0NzExMS1kMzE0LTQ3OTktODcyZC05MDhmZGM3YWFmY2YiLCJwcmV2aWV3TmFtZSI6IjQucG5nIiwibGljZW5zZVN0YXR1cyI6Ik9LIiwic2VydmljZSI6IiIsImlzVGVtcGxhdGUiOiIiLCJ3aWR0aCI6IjgxMSIsInNpbXBsZVZpZXdlciI6ZmFsc2UsImxhc3RNb2RpZmllZCI6MTY5NDA4Nzc0MTAwMCwiZXhjZWVkUGFnZVdpZHRoIjpmYWxzZSwib0NsaWVudElkIjoiIn0=
Delete Branch
Delete branch is very simple, just delete the directory of the specified branch directly.
Merge Branch To Main
Merge a branch into the main branch can be divided into two steps:
Delete all the snapshots, tags and schemas in the main branch that are created after the created tag for the branch
Copy snapshots, tags and schemas from the branch to the main branch.
The merged branch can still be read and written by jobs and the data in the branch is still independent from the main branch.
eyJleHRTcnZJbnRlZ1R5cGUiOiIiLCJnQ2xpZW50SWQiOiIiLCJjcmVhdG9yTmFtZSI6IkZhbmcgWW9uZyIsIm91dHB1dFR5cGUiOiJibG9jayIsImxhc3RNb2RpZmllck5hbWUiOiJGYW5nIFlvbmciLCJsYW5ndWFnZSI6ImVuIiwiZGlhZ3JhbURpc3BsYXlOYW1lIjoiIiwic0ZpbGVJZCI6IiIsImF0dElkIjoiMjYzNDI4NjU2IiwiZGlhZ3JhbU5hbWUiOiLmnKrlkb3lkI3nu5jlm74iLCJhc3BlY3QiOiIiLCJsaW5rcyI6ImF1dG8iLCJjZW9OYW1lIjoiUElQLTk6IFN1cHBvcnQgQnJhbmNoIiwidGJzdHlsZSI6InRvcCIsImNhbkNvbW1lbnQiOmZhbHNlLCJkaWFncmFtVXJsIjoiIiwiY3N2RmlsZVVybCI6IiIsImJvcmRlciI6dHJ1ZSwibWF4U2NhbGUiOiIxIiwib3duaW5nUGFnZUlkIjoyNjM0MjczODMsImVkaXRhYmxlIjpmYWxzZSwiY2VvSWQiOjI2MzQyODY2MiwicGFnZUlkIjoiIiwibGJveCI6dHJ1ZSwic2VydmVyQ29uZmlnIjp7ImVtYWlscHJldmlldyI6IjEifSwib2RyaXZlSWQiOiIiLCJyZXZpc2lvbiI6MSwibWFjcm9JZCI6IjMyOTA5NTEzLWU0OWMtNDA2MC1hZjhiLTE4NDAwOGI2ZjY3ZiIsInByZXZpZXdOYW1lIjoi5pyq5ZG95ZCN57uY5Zu+LnBuZyIsImxpY2Vuc2VTdGF0dXMiOiJPSyIsInNlcnZpY2UiOiIiLCJpc1RlbXBsYXRlIjoiIiwid2lkdGgiOiI2MDAiLCJzaW1wbGVWaWV3ZXIiOmZhbHNlLCJsYXN0TW9kaWZpZWQiOjE2OTQwODc3ODcwMDAsImV4Y2VlZFBhZ2VXaWR0aCI6ZmFsc2UsIm9DbGllbnRJZCI6IiJ9
Replace Main With Branch
We need to support replacing the main branch with a branch without affecting streaming and batch data read and write on the branch. To achieve this, we need to do the following steps:
Calculate the snapshots, tags and schemas which should be copied from the main branch to target branch
Update the Main Branch File
to the target branch
Drop the previous main branch, including snapshots, tags and schemas.
After the above steps, the main branch will be replaced with the target branch and the existing jobs can still read and write data in the branch.
eyJleHRTcnZJbnRlZ1R5cGUiOiIiLCJnQ2xpZW50SWQiOiIiLCJjcmVhdG9yTmFtZSI6IkZhbmcgWW9uZyIsIm91dHB1dFR5cGUiOiJibG9jayIsImxhc3RNb2RpZmllck5hbWUiOiJGYW5nIFlvbmciLCJsYW5ndWFnZSI6ImVuIiwiZGlhZ3JhbURpc3BsYXlOYW1lIjoiIiwic0ZpbGVJZCI6IiIsImF0dElkIjoiMjYzNDI4NjU4IiwiZGlhZ3JhbU5hbWUiOiI2IiwiYXNwZWN0IjoiIiwibGlua3MiOiJhdXRvIiwiY2VvTmFtZSI6IlBJUC05OiBTdXBwb3J0IEJyYW5jaCIsInRic3R5bGUiOiJ0b3AiLCJjYW5Db21tZW50IjpmYWxzZSwiZGlhZ3JhbVVybCI6IiIsImNzdkZpbGVVcmwiOiIiLCJib3JkZXIiOnRydWUsIm1heFNjYWxlIjoiMSIsIm93bmluZ1BhZ2VJZCI6MjYzNDI3MzgzLCJlZGl0YWJsZSI6ZmFsc2UsImNlb0lkIjoyNjM0Mjg2NjIsInBhZ2VJZCI6IiIsImxib3giOnRydWUsInNlcnZlckNvbmZpZyI6eyJlbWFpbHByZXZpZXciOiIxIn0sIm9kcml2ZUlkIjoiIiwicmV2aXNpb24iOjEsIm1hY3JvSWQiOiI0ZDVmYTFjOS1mZTkxLTQzNDctYmI5Yy02MmRlZTFjMTVhMWIiLCJwcmV2aWV3TmFtZSI6IjYucG5nIiwibGljZW5zZVN0YXR1cyI6Ik9LIiwic2VydmljZSI6IiIsImlzVGVtcGxhdGUiOiIiLCJ3aWR0aCI6IjYwMCIsInNpbXBsZVZpZXdlciI6ZmFsc2UsImxhc3RNb2RpZmllZCI6MTY5NDA4NzgyNTAwMCwiZXhjZWVkUGFnZVdpZHRoIjpmYWxzZSwib0NsaWVudElkIjoiIn0=