static void applySchemaUpdates(Table table, SchemaUpdate.Consumer updates) {
    if (updates == null || updates.empty()) {
      // no updates to apply
      return;
    }

    Tasks.range(1)
        .retry(IcebergSinkConfig.SCHEMA_UPDATE_RETRIES)
        .run(notUsed -> commitSchemaUpdates(table, updates));
  }
@Test
  public void testApplyNestedSchemaUpdates() {
    UpdateSchema updateSchema = mock(UpdateSchema.class);
    Table table = mock(Table.class);
    when(table.schema()).thenReturn(NESTED_SCHEMA);
    when(table.updateSchema()).thenReturn(updateSchema);

    // the updates to "st.i" should be ignored as it already exists and is the same type
    SchemaUpdate.Consumer consumer = new SchemaUpdate.Consumer();
    consumer.addColumn("st", "i", IntegerType.get());
    consumer.updateType("st.i", IntegerType.get());
    consumer.makeOptional("st.i");
    consumer.updateType("st.f", DoubleType.get());
    consumer.addColumn("st", "s", StringType.get());

    SchemaUtils.applySchemaUpdates(table, consumer);
    verify(table).refresh();
    verify(table).updateSchema();

    verify(updateSchema).addColumn(eq("st"), eq("s"), isA(StringType.class));
    verify(updateSchema).updateColumn(eq("st.f"), isA(DoubleType.class));
    verify(updateSchema).makeColumnOptional(eq("st.i"));
    verify(updateSchema).commit();

    // check that there are no unexpected invocations...
    verify(updateSchema).addColumn(anyString(), anyString(), any());
    verify(updateSchema).updateColumn(any(), any());
    verify(updateSchema).makeColumnOptional(any());
  }