public void updateTableSchema(String tableName, Schema schema, List<String> partitionFields) {
    Table existingTable = bigquery.getTable(TableId.of(projectId, datasetName, tableName));
    ExternalTableDefinition definition = existingTable.getDefinition();
    Schema remoteTableSchema = definition.getSchema();
    List<Field> finalTableFields = new ArrayList<>(schema.getFields());
    // Add the partition fields into the schema to avoid conflicts while updating. And ensure the partition fields are at the end to
    // avoid unnecessary updates.
    List<Field> bqPartitionFields = remoteTableSchema.getFields().stream()
        .filter(field -> partitionFields.contains(field.getName()))
        .collect(Collectors.toList());
    finalTableFields.addAll(bqPartitionFields);
    Schema finalSchema = Schema.of(finalTableFields);
    boolean sameSchema = definition.getSchema() != null && definition.getSchema().equals(finalSchema);
    boolean samePartitionFilter = partitionFields.isEmpty()
        || (requirePartitionFilter == (definition.getHivePartitioningOptions().getRequirePartitionFilter() != null && definition.getHivePartitioningOptions().getRequirePartitionFilter()));
    if (sameSchema && samePartitionFilter) {
      LOG.info("No table update is needed.");
      return; // No need to update schema.
    }
    if (!StringUtils.isNullOrEmpty(bigLakeConnectionId)) {
      Table updatedTable =
              existingTable.toBuilder().setDefinition(StandardTableDefinition.of(finalSchema)).build();
      updatedTable.update();
    } else {
      ExternalTableDefinition.Builder builder = definition.toBuilder();
      builder.setSchema(finalSchema);
      builder.setAutodetect(false);
      if (definition.getHivePartitioningOptions() != null) {
        builder.setHivePartitioningOptions(definition.getHivePartitioningOptions().toBuilder().setRequirePartitionFilter(requirePartitionFilter).build());
      }
      Table updatedTable = existingTable.toBuilder()
              .setDefinition(builder.build())
              .build();
      bigquery.update(updatedTable);
    }
  }
@Test
  void skipUpdatingSchema_partitioned() throws Exception {
    BigQuerySyncConfig config = new BigQuerySyncConfig(properties);
    client = new HoodieBigQuerySyncClient(config, mockBigQuery);
    Table mockTable = mock(Table.class);
    ExternalTableDefinition mockTableDefinition = mock(ExternalTableDefinition.class);
    // The table schema has no change: it contains a "field" and a "partition_field".
    Schema schema = Schema.of(Field.of("field", StandardSQLTypeName.STRING));
    List<String> partitionFields = new ArrayList<String>();
    partitionFields.add("partition_field");
    List<Field> bqFields = new ArrayList<Field>();
    // The "partition_field" always follows "field".
    bqFields.add(Field.of("field", StandardSQLTypeName.STRING));
    bqFields.add(Field.of("partition_field", StandardSQLTypeName.STRING));
    Schema bqSchema = Schema.of(bqFields);
    HivePartitioningOptions hivePartitioningOptions = HivePartitioningOptions.newBuilder().setRequirePartitionFilter(true).build();

    when(mockBigQuery.getTable(any())).thenReturn(mockTable);
    when(mockTable.getDefinition()).thenReturn(mockTableDefinition);
    when(mockTableDefinition.getSchema()).thenReturn(bqSchema);
    when(mockTableDefinition.getHivePartitioningOptions()).thenReturn(hivePartitioningOptions);

    client.updateTableSchema(TEST_TABLE, schema, partitionFields);
    // Expect no update.
    verify(mockBigQuery, never()).update(mockTable);
  }