@Override
  public BeamSqlTable buildBeamSqlTable(Table tableDefinition) {
    ObjectNode tableProperties = tableDefinition.getProperties();

    try {
      RowJson.RowJsonDeserializer deserializer =
          RowJson.RowJsonDeserializer.forSchema(getSchemaIOProvider().configurationSchema())
              .withNullBehavior(RowJson.RowJsonDeserializer.NullBehavior.ACCEPT_MISSING_OR_NULL);

      Row configurationRow =
          newObjectMapperWith(deserializer).readValue(tableProperties.toString(), Row.class);

      SchemaIO schemaIO =
          getSchemaIOProvider()
              .from(tableDefinition.getLocation(), configurationRow, tableDefinition.getSchema());

      return new SchemaIOTableWrapper(schemaIO);
    } catch (InvalidConfigurationException | InvalidSchemaException e) {
      throw new InvalidTableException(e.getMessage());
    } catch (JsonProcessingException e) {
      throw new AssertionError("Failed to re-parse TBLPROPERTIES JSON " + tableProperties);
    }
  }
@Test
  public void testBuildIOReader_withProjectionPushdown() {
    TestSchemaIOTableProviderWrapper provider = new TestSchemaIOTableProviderWrapper();
    BeamSqlTable beamSqlTable = provider.buildBeamSqlTable(testTable);

    PCollection<Row> result =
        beamSqlTable.buildIOReader(
            pipeline.begin(),
            new DefaultTableFilter(ImmutableList.of()),
            ImmutableList.of("f_long"));
    Schema outputSchema = Schema.builder().addInt64Field("f_long").build();
    PAssert.that(result)
        .containsInAnyOrder(
            Row.withSchema(outputSchema).addValues(0L).build(),
            Row.withSchema(outputSchema).addValues(1L).build());

    pipeline.run();
  }