SNOW-1729292 modify iceberg tree based on record data

[sfc-gh-bzabek]

Overview

SNOW-1729292

There is still a lot work to be done. Tell me if you find this approach good.

1. Ingest-sdk returns list of columnsToEvolve inside an InsertError
2. get schemas from channel for these columns
3. get column schemas from record
4. resolve which column is new

• if column from record is not present in channels schema than it's new
• add this column

5. resolve which column is modified

• if the column is present in a channel it means it's not new - hence it's to be modified
• merge column from channel with a column from record
• set data type

The whole logic responsible for iceberg schema evolution is package private. Only IcebergSchemaEvolutionService is public.

It doesn't work with ingest-sdk 3.0.0, local jar needs to be build from ingest-sdk master.

TODO:

• I didn't inspect if columns nullability evolution is handled
• column comments are not handled
• tests for potential SQL injection (I think we need another Jira for it)

Pre-review checklist

• This change should be part of a Behavior Change Release. See go/behavior-change.
• This change has passed Merge gate tests
• Snowpipe Changes
• Snowpipe Streaming Changes
• This change is TEST-ONLY
• This change is README/Javadocs only
• This change is protected by a config parameter <PARAMETER_NAME> eg snowflake.ingestion.method.
  • Yes - Added end to end and Unit Tests.
  • No - Suggest why it is not param protected
• Is his change protected by parameter <PARAMETER_NAME> on the server side?
  • The parameter/feature is not yet active in production (partial rollout or PrPr, see Changes for Unreleased Features and Fixes).
  • If there is an issue, it can be safely mitigated by turning the parameter off. This is also verified by a test (See go/ppp).

[sfc-gh-mbobowski]

Actually it is not used anymore. You can safely delete it.

[sfc-gh-bzabek]

I'm confused. I use icebergTableSchemaResolver a couple of times. (?)

[sfc-gh-mbobowski]

I mean the constructor below marked with @VisibleForTesting (not affected by your changes).

[sfc-gh-bzabek]

This class was moved to a correct module.

[sfc-gh-mbobowski]

We have two separated flows: JSON without schema and AVRO/Protobuf.

I would extract fromConnectSchema to a separate class SchematizedIcebergColumnTreeFactory and rename this class to NoSchemaIcebergColumnTreeFactory. WDYT?

[sfc-gh-bzabek]

Hmm, so what we have:

• from Iceberg (channel) schema - it's used before both workflows
• from json payload - workflow without schema
• from record schema - now there is only fromConnectSchema method. Do we have to also write seperate logic to parse avro and protobuf schema? I think not because Converter will parse it into connect schema when we ancounter AVRO or Protobuf. (For sure we ust test it)

When I wrote the factory, I thought it will be a bit over engineering to split it. However if we are going to need more methods then sure. Logically we have 3 parts.

[sfc-gh-mbobowski]

There is no need to iterate twice over the list. The columns from modifiedOrAddedColumns are either modified or added, right?

[sfc-gh-bzabek]

Righ, however I found my approach simpler and didn't care very much about performance of that.

Create 2 lists and write and if to put an element into one or another list.

  private Pair<List<IcebergColumnTree>, List<IcebergColumnTree>> distinguish(
          List<IcebergColumnTree> alreadyExistingColumns,
          List<IcebergColumnTree> modifiedOrAddedColumns) {

    ArrayList<IcebergColumnTree> columnsToModify = new ArrayList<>();
    ArrayList<IcebergColumnTree> columnsToAdd = new ArrayList<>();
    
    for (IcebergColumnTree tree : modifiedOrAddedColumns) {
      if (alreadyExistingColumns.stream()
              .anyMatch(alreadyExisting -> alreadyExisting.getColumnName()
                      .equalsIgnoreCase(tree.getColumnName()))) {
        
        columnsToModify.add(tree);
      } else {
        columnsToAdd.add(tree);
      }
    }
    return Pair<List<IcebergColumnTree>, List<IcebergColumnTree>>(columnsToAdd, columnsToModify);
  }

But there is not "Pair" in JDK8.
This approach is a bit more messy. Slightly more performant.

(It's messy ->) The second alternative I see is to distinguish columnsToAdd, and then having columnsToAdd iterate over columns to evolve and again match them with modifiedOrAddedColumns list.

[sfc-gh-bzabek]

None is perfect.

[sfc-gh-mbobowski]

For me it's a matter of avoiding code duplication rather than optimizing the code execution time. I don't see any problem with creating a wrapper class for two lists.
Anyway it's not a blocker to me.

[sfc-gh-mbobowski]

It would be easier to convert alreadyExistingColumns from List to Set.

[sfc-gh-bzabek]

I can use Set everywhere instead of a list.

[sfc-gh-bzabek]

But for IcebergColumnTree I didn't implement equals nor hashCode. I don't feel it.

[sfc-gh-mbobowski]

Good point. Is there any problem with implementing these methods or the final code doesn't look cleaner?

[sfc-gh-bzabek]

If we implement only for columnName - there is not a problem. Using it for both doesn't make sense.
I don't think using Set will change anything in a logic. It may (shouldn't) sneakily replace a column somewhere... I can change it, never say never.

[sfc-gh-mbobowski]

it can be private

[sfc-gh-bzabek]

Let's keep it consise with others. I don't want to restrict it's usage.

[sfc-gh-mbobowski]

Let's move it at the beginning of the class.

[sfc-gh-bzabek]

Done

[sfc-gh-mbobowski]

I still see some space for minor improvements, but I am ok with merging it at this point.

[sfc-gh-wtrefon]

nice

[sfc-gh-wtrefon]

Looking much better now, really like how you split the IcebergFieldNode class. Good job!