class TablesOperationTransformer extends OperationTransformer<TableMetadataUnit> {

  public List<Overwrite> toOverwrite(List<TableMetadataUnit> units) {
    Map<TableKey, Map<String, List<TableMetadataUnit>>> data = units.stream()
      .collect(
        Collectors.groupingBy(TableKey::of,
          Collectors.groupingBy(TableMetadataUnit::metadataKey)
        ));

    return data.entrySet().parallelStream()
      .map(dataEntry -> dataEntry.getValue().entrySet().parallelStream()
        .map(operationEntry -> {
          TableKey tableKey = dataEntry.getKey();

          String location = tableKey.toLocation(context.table().location());

          Map<String, Object> filterConditions = new HashMap<>(tableKey.toFilterConditions());
          filterConditions.put(IcebergTables.METADATA_KEY, operationEntry.getKey());
          Expression expression = context.transformer().filter().transform(filterConditions);

          return toOverwrite(location, expression, operationEntry.getValue());
        })
        .collect(Collectors.toList()))
      .flatMap(Collection::stream)
      .collect(Collectors.toList());
  }

  public  TablesOperationTransformer(MetastoreContext<TableMetadataUnit> context);

}