Flink: Use Tag or Branch to scan data.

[hililiwei]

What is the purpose of the change

Scan data using a specified tag or branch.

Brief change log

Add the following syntax
SQL:

* SELECT * FROM sample /*+ OPTIONS('tag'='t1')*/ ;
* SELECT * FROM sample /*+ OPTIONS('branch'='t1')*/ ;

API:

    public Builder tag(String tag) {
      contextBuilder.useTag(tag);
      return this;
    }

    public Builder branch(String branch) {
      contextBuilder.useBranch(branch);
      return this;
    }
    public Builder startTag(String startTag) {
      contextBuilder.startTag(startTag);
      return this;
    }

    public Builder endTag(String endTag) {
      contextBuilder.endTag(endTag);
      return this;
    }

cc @amogh-jahagirdar

[amogh-jahagirdar]

This has a dependency on https://github.com/apache/iceberg/pull/4428/files, I will follow up on that PR.

[hililiwei]

As the core is being modified, this PR needs to wait for #5364 and #5475 to complete.

[hililiwei]

cc @amogh-jahagirdar @stevenzwu

[hililiwei]

cc @amogh-jahagirdar @stevenzwu @rdblue, could you please take a look when you are available?

[amogh-jahagirdar]

Thanks for this @hililiwei ! left some comments/questions it makes sense to just carry this PR forward based on our discussion in #6660

[amogh-jahagirdar]

I think it'll help if at some point (probably options/conf default) we set the default branch to main and always assume a branch is set in this code. Then we can just do table.snapshot(branch).

Nit on log message:

"No snapshots on branch %s in table %s"

[jackye1995]

+1, can we actually do that in this PR? (can we check what is done in Spark, if we decided to go with using main branch or using if else? I cannot remember now)

[hililiwei]

spark doesn't seem to use Main as the default.

My main concern is that main is not compatible with older version iceberg table.

[amogh-jahagirdar]

When we read the table metadata with the current Iceberg version we will set the main branch. https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/TableMetadataParser.java#L444 so it will be compatible with older version of tables.

[amogh-jahagirdar]

Also sorry if I missed it, but at some point shouldn't there be validation that the start tag is an ancestor of the end tag?

[hililiwei]

They are eventually converted to snapshot id and checked during the planTasks phase

[jackye1995]

@stevenzwu since you are reviewing #6660, could you also take a look at this?

[hililiwei]

#5984 seems to be the prerequisite for flink to implement stream incremental read with branch.

[jackye1995]

nit: cannot both be set?

[jackye1995]

Mostly looks good to me, just 1 nit comment

[hililiwei]

@stevenzwu addressed the round of comments, PTAL, thx.

[stevenzwu]

nit: this can be appendToTable(null, records), right?

[hililiwei]

Currently it won't work because we have two methods:
appendToTable(String branch, List<Record> records)
and
appendToTable(StructLike partition, List<Record> records)

Unless I rearrange the order of branch and place it after records, but I prefer to leave it as it is currently, at the front.

[amogh-jahagirdar]

Looks good to me, thanks @hililiwei for the contribution!

[jackye1995]

Looks like all comments are addressed and we have enough votes, thanks for the work @hililiwei , and thanks for the review @stevenzwu and @amogh-jahagirdar !

[hililiwei]

Thanks for the review @jackye1995 @stevenzwu @amogh-jahagirdar