Schema getTableSchema(HoodieTableMetaClient metaClient, List<String> partitionFields) {
    try {
      Schema schema = convertSchema(tableSchemaResolverSupplier.apply(metaClient).getTableAvroSchema());
      if (partitionFields.isEmpty()) {
        return schema;
      } else {
        return Schema.of(schema.getFields().stream().filter(field -> !partitionFields.contains(field.getName())).collect(Collectors.toList()));
      }
    } catch (Exception e) {
      throw new HoodieBigQuerySyncException("Failed to get table schema", e);
    }
  }
@Test
  void getTableSchema_withPartitionFields() throws Exception {
    HoodieTableMetaClient mockMetaClient = mock(HoodieTableMetaClient.class);
    TableSchemaResolver mockTableSchemaResolver = mock(TableSchemaResolver.class);
    when(mockTableSchemaResolver.getTableAvroSchema()).thenReturn(PRIMITIVE_TYPES);
    BigQuerySchemaResolver resolver = new BigQuerySchemaResolver(metaClient -> mockTableSchemaResolver);

    com.google.cloud.bigquery.Schema expected = com.google.cloud.bigquery.Schema.of(
        Field.newBuilder("requiredBoolean", StandardSQLTypeName.BOOL).setMode(Field.Mode.REQUIRED).build(),
        Field.newBuilder("optionalBoolean", StandardSQLTypeName.BOOL).setMode(Field.Mode.NULLABLE).build(),
        Field.newBuilder("requiredInt", StandardSQLTypeName.INT64).setMode(Field.Mode.REQUIRED).build(),
        Field.newBuilder("optionalInt", StandardSQLTypeName.INT64).setMode(Field.Mode.NULLABLE).build(),
        Field.newBuilder("requiredLong", StandardSQLTypeName.INT64).setMode(Field.Mode.REQUIRED).build(),
        Field.newBuilder("optionalLong", StandardSQLTypeName.INT64).setMode(Field.Mode.NULLABLE).build(),
        Field.newBuilder("requiredDouble", StandardSQLTypeName.FLOAT64).setMode(Field.Mode.REQUIRED).build(),
        Field.newBuilder("optionalDouble", StandardSQLTypeName.FLOAT64).setMode(Field.Mode.NULLABLE).build(),
        Field.newBuilder("requiredFloat", StandardSQLTypeName.FLOAT64).setMode(Field.Mode.REQUIRED).build(),
        Field.newBuilder("optionalFloat", StandardSQLTypeName.FLOAT64).setMode(Field.Mode.NULLABLE).build(),
        Field.newBuilder("optionalString", StandardSQLTypeName.STRING).setMode(Field.Mode.NULLABLE).build(),
        Field.newBuilder("requiredBytes", StandardSQLTypeName.BYTES).setMode(Field.Mode.REQUIRED).build(),
        Field.newBuilder("optionalBytes", StandardSQLTypeName.BYTES).setMode(Field.Mode.NULLABLE).build(),
        Field.newBuilder("requiredEnum", StandardSQLTypeName.STRING).setMode(Field.Mode.REQUIRED).build(),
        Field.newBuilder("optionalEnum", StandardSQLTypeName.STRING).setMode(Field.Mode.NULLABLE).build());

    // expect 'requiredString' field to be removed
    Assertions.assertEquals(expected, resolver.getTableSchema(mockMetaClient, Collections.singletonList("requiredString")));
  }