[core] Support reading partition from fallback branch when not found in current branch

[tsreaper]

Purpose

In this PR we have supported reading partitions from fallback branch.

You can set the table option scan.fallback-branch so that when a batch job reads from the current branch, if a partition does not exist, the reader will try to read this partition from the fallback branch. For streaming read jobs, this feature is currently not supported, and will only produce results from the current branch.

What's the use case of this feature? Say you have created a Paimon table partitioned by date. You have a long-running streaming job which inserts records into Paimon, so that today's data can be queried in time. You also have a batch job which runs at every night to insert corrected records of yesterday into Paimon, so that the preciseness of the data can be promised.

When you query from this Paimon table, you would like to first read from the results of batch job. But if a partition (for example, today's partition) does not exist in its result, then you would like to read from the results of streaming job. In this case, you can create a branch for streaming job, and set scan.fallback-branch to this streaming branch.

Let's look at an example.

-- create Paimon table
CREATE TABLE T (
    dt STRING NOT NULL,
    name STRING NOT NULL,
    amount BIGINT
) PARTITIONED BY (dt);

-- create a branch for streaming job
CALL sys.create_branch('default.T', 'test');

-- set primary key and bucket number for the branch
ALTER TABLE `T$branch_test` SET (
    'primary-key' = 'dt,name',
    'bucket' = '2'
);

-- set fallback branch
ALTER TABLE T SET (
    'scan.fallback-branch' = 'test'
);

-- set changelog producer for the streaming branch, in case a streaming job would like to read from it in the future
ALTER TABLE `T$branch_test` SET (
    'changelog-producer' = 'lookup'
);

-- write records into the streaming branch
INSERT INTO `T$branch_test` VALUES ('20240725', 'apple', 4), ('20240725', 'peach', 10), ('20240726', 'cherry', 3), ('20240726', 'pear', 6);

-- write records into the default branch
INSERT INTO T VALUES ('20240725', 'apple', 5), ('20240725', 'banana', 7);

SELECT * FROM T;
/*
+------------------+------------------+--------+
|               dt |             name | amount |
+------------------+------------------+--------+
|         20240725 |            apple |      5 |
|         20240725 |           banana |      7 |
|         20240726 |           cherry |      3 |
|         20240726 |             pear |      6 |
+------------------+------------------+--------+
*/

-- reset fallback branch
ALTER TABLE T RESET ( 'scan.fallback-branch' );

-- now it only reads from default branch
SELECT * FROM T;
/*
+------------------+------------------+--------+
|               dt |             name | amount |
+------------------+------------------+--------+
|         20240725 |            apple |      5 |
|         20240725 |           banana |      7 |
+------------------+------------------+--------+
*/

Tests

IT cases.

API and Format

No changes in format.

Documentation

Yes. Document is also updated.

[JingGe]

Thanks for driving it. Just left some comments, PTAL

[JingGe]

What would be the expected behaviour if the feature for the streaming read had correctly implemented?

[tsreaper]

We haven't decided yet. We need to talk with the users about it.

[JingGe]

better first run the query without setting scan.fallback-branch and return the two rows of 20240725, then set 'scan.fallback-branch' = 'test' and run the query again to return the result of 4 rows with the data from the fallback branch. WDYT?

[JingGe]

How to reset it? 'scan.fallback-branch' = null ?

[tsreaper]

ALTER TABLE T RESET ('scan.fallback-branch'), see Paimon docs.

[JingGe]

Does it make sense to move the two Preconditions checks to the constructor of FallbackReadFileStoreTable ?

[JingGe]

Thanks for driving it. Just left some comments, PTAL

[JingGe]

Nit: additional test when scan.fallback-branch is not set, i.e. only return data from main branch

[JingsongLi]

With the subsequent development, different branches may have significant differences and require a lot of options modifications. Can we consider directly supporting the modification of all options of an empty table?

[JingsongLi]

put two alter together.

[JingsongLi]

You can introduce private method to createNoWrappedTable.

[JingsongLi]

+1