Add support for Flink Beam templates

[jogrogan]

Several changes wrapped under this:

1. Add new FlinkBeam enum type to SqlJobs crd & regenerate model files
2. Pass source information (database/schema/table) through Job type to allow to pop out in job templates
3. Add template equality checks to only match a given template if a specified condition is met
   {{field==value}} or {{field!=value}}
4. Add new fieldMap logic to generate a mapping of fields from source to destination.
   Logic only applies when the query is "trivial", meaning Projects & Table Scans only. If there is a use case this can be extended to joins in the future but would require prefixing fields in the field map with the relevant source
5. Add lazy evaluation of sql & fieldmap functions & allow them to throw SqlExceptions
6. Lots of testing - found some quoting and newline issues when applying yaml and fixed

Field map testing and examples:

1. Standard select query

create or replace materialized view ads.pages as select page_urn from ads.page_views;
"PAGE_URN" -> "PAGE_URN"

2. Select * query

create or replace materialized view ads.pages as select * from ads.page_views;
"MEMBER_URN" -> "MEMBER_URN"
"PAGE_URN" -> "PAGE_URN"

3. Select query with alias

create or replace materialized view ads.pages as select page_urn as new_page_urn from ads.page_views;
"PAGE_URN" -> "NEW_PAGE_URN"

4. Mulitple fields and aliases

create or replace materialized view ads."PAGE_VIEWS$filter" as select first_name as page_urn, last_name as member_urn from profile.members;
"LAST_NAME" -> "MEMBER_URN"
"FIRST_NAME" -> "PAGE_URN"

5. Implicit aliasing

create or replace materialized view "VENICE"."test-store$insert-partial" ("KEY_id", "intField") as select "KEY", "intField" from "VENICE"."test-store-primitive";
"intField" -> "intField"
"KEY" -> "KEY_id"

6. Nested fields

create or replace materialized view "VENICE"."test-store$insert-nested" ("KEY_id", "intField") as select "KEY", "nestedField"['nestedIntField'] from "VENICE"."test-store-primitive";
"nestedField.nestedIntField" -> "intField",
"KEY" -> "KEY_id"

Error cases:

1. Aliasing a literal

create or replace materialized view ads.pages as select page_urn, '1' as new_field from ads.page_views;
Field mapping is only supported for trivial queries with simple projections and aliasing.

2. Unknown source field

create or replace materialized view ads."PAGE_VIEWS$myview" as select "unknown" from profile.members;
Column 'unknown' not found in any table

3. Unknown sink field

create or replace materialized view ads."PAGE_VIEWS$myview" as select first_name from profile.members;
ERROR Field FIRST_NAME not found in sink schema

4. Non-trivial query (join)

create or replace materialized view ads.audience as select first_name, last_name from ads.page_views natural join profile.members;
Field mapping is only supported for trivial queries with simple projections and aliasing.

Added more tests for deeply nested fields, aliasing nested fields, etc.

[ryannedolan]

This is a little funky. Why not have methods query(dialect), sql(dialect), and fieldMap() that return this data? They could still be supplied to an Environment and evaluated lazily, right?

[jogrogan]

Part of it is a package-problem. Job is in hoptimator-api which has no dependencies. It can't make the function calls directly, lots of local info is contained within PipelineRel where the functions exist.

I also did not want to have an expanding list of arguments in the Job constructor for each unique function.

I can remove the eval function and just have query(dialect), sql(dialect), and fieldMap(dialect) as you suggested, I do think that's better as the calls are less "magic"

[jogrogan]

Actually, even AnsiSqlDialect exists within Calcite, so unless we add that as a dependency, these have to remain without an argument and we'll still need that wrap function.

But I can have query(), sql(), and fieldMap() to remove the magic string bit

[ryannedolan]

super cool

[ryannedolan]

This is still weird, but at least it's private now.