Use SerializableTable instead of TableLoader in Flink

[aokolnychyi]

I am wondering whether we can in certain cases switch to using SerializableTable instead of TableLoader in Flink. In particular, I am talking about FlinkInputFormat. My primary motivation is the ability to pass just a Table object instead of a number of individual parts. For example, I want to pass Table to PartitionUtil#constantsMap as the partition type depends on all specs. A side benefit of using SerializableTable is that we won't have to load and close the catalog.

[aokolnychyi]

Thoughts, @openinx @stevenzwu?

[aokolnychyi]

There may be still valid places to use TableLoader, though. I think the initial loading in FlinkSource can be done through it.

[stevenzwu]

I agree that we can switch to SerializableTable in some cases (like FlinkInputFormat you mentioned). Since FlinkSource already loaded the table, it can just pass in a SerializableTable copy. We can also simplify the params for FlinkInputFormat constructor

[aokolnychyi]

Okay, I'll fire a PR to discuss it in detail.

[github-actions]

This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.

[github-actions]

This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'