public static List<String> generateEvolutionDDL(String stagingTableName,
      String finalTableName,
      Optional<String> optionalStagingDbName,
      Optional<String> optionalFinalDbName,
      Schema evolvedSchema,
      boolean isEvolutionEnabled,
      Map<String, String> evolvedColumns,
      Optional<Table> destinationTableMeta) {
    // If schema evolution is disabled, then do nothing OR
    // If destination table does not exists, then do nothing
    if (!isEvolutionEnabled || !destinationTableMeta.isPresent()) {
      return Collections.emptyList();
    }

    String stagingDbName = optionalStagingDbName.isPresent() ? optionalStagingDbName.get() : DEFAULT_DB_NAME;
    String finalDbName = optionalFinalDbName.isPresent() ? optionalFinalDbName.get() : DEFAULT_DB_NAME;

    List<String> ddl = Lists.newArrayList();

    // Evolve schema
    Table destinationTable = destinationTableMeta.get();
    if (destinationTable.getSd().getCols().size() == 0) {
      log.warn("Desination Table: " + destinationTable + " does not has column details in StorageDescriptor. "
          + "It is probably of Avro type. Cannot evolve via traditional HQL, so skipping evolution checks.");
      return ddl;
    }
    for (Map.Entry<String, String> evolvedColumn : evolvedColumns.entrySet()) {
      // Find evolved column in destination table
      boolean found = false;
      for (FieldSchema destinationField : destinationTable.getSd().getCols()) {
        if (destinationField.getName().equalsIgnoreCase(evolvedColumn.getKey())) {
          // If evolved column is found, but type is evolved - evolve it
          // .. if incompatible, isTypeEvolved will throw an exception
          boolean typeEvolved;
          try {
            typeEvolved = isTypeEvolved(evolvedColumn.getValue(), destinationField.getType());
          } catch (Exception e) {
            throw new RuntimeException(
                String.format("Unable to evolve schema for table %s.%s", finalDbName, finalTableName), e);
          }
          if (typeEvolved) {
            ddl.add(String.format("USE %s%n", finalDbName));
            ddl.add(String.format("ALTER TABLE `%s` CHANGE COLUMN `%s` `%s` %s COMMENT '%s'",
                finalTableName, evolvedColumn.getKey(), evolvedColumn.getKey(), evolvedColumn.getValue(),
                escapeStringForHive(destinationField.getComment())));
          }
          found = true;
          break;
        }
      }
      if (!found) {
        // If evolved column is not found ie. its new, add this column
        String flattenSource = evolvedSchema.getField(evolvedColumn.getKey()).getProp("flatten_source");
        if (StringUtils.isBlank(flattenSource)) {
          flattenSource = evolvedSchema.getField(evolvedColumn.getKey()).name();
        }
        // Note: Hive does not support fully qualified Hive table names such as db.table for ALTER TABLE in v0.13
        // .. hence specifying 'use dbName' as a precursor to rename
        // Refer: HIVE-2496
        ddl.add(String.format("USE %s%n", finalDbName));
        ddl.add(String.format("ALTER TABLE `%s` ADD COLUMNS (`%s` %s COMMENT 'from flatten_source %s')",
            finalTableName, evolvedColumn.getKey(), evolvedColumn.getValue(), flattenSource));
      }
    }

    return ddl;
  }
public static List < String > generateEvolutionDDL ( String stagingTableName , String finalTableName , Optional < String > optionalStagingDbName , Optional < String > optionalFinalDbName , Schema evolvedSchema , boolean isEvolutionEnabled , Map < String , String > evolvedColumns , Optional < Table > destinationTableMeta ) { if ( ! isEvolutionEnabled || ! destinationTableMeta . isPresent ( ) ) { return Collections . emptyList ( ) ; } String stagingDbName = optionalStagingDbName . isPresent ( ) ? optionalStagingDbName . get ( ) : DEFAULT_DB_NAME ; String finalDbName = optionalFinalDbName . isPresent ( ) ? optionalFinalDbName . get ( ) : DEFAULT_DB_NAME ; List < String > ddl = Lists . newArrayList ( ) ; Table destinationTable = destinationTableMeta . get ( ) ; if ( destinationTable . getSd ( ) . getCols ( ) . size ( ) == 0 ) { log . warn ( "Desination Table: " + destinationTable + " does not has column details in StorageDescriptor. " + "It is probably of Avro type. Cannot evolve via traditional HQL, so skipping evolution checks." ) ; return ddl ; } for ( Map . Entry < String , String > evolvedColumn : evolvedColumns . entrySet ( ) ) { boolean found = false ; for ( FieldSchema destinationField : destinationTable . getSd ( ) . getCols ( ) ) { if ( destinationField . getName ( ) . equalsIgnoreCase ( evolvedColumn . getKey ( ) ) ) { boolean typeEvolved ; try { typeEvolved = isTypeEvolved ( evolvedColumn . getValue ( ) , destinationField . getType ( ) ) ; } catch ( Exception e ) { throw new RuntimeException ( String . format ( "Unable to evolve schema for table %s.%s" , finalDbName , finalTableName ) , e ) ; } if ( typeEvolved ) { ddl . add ( String . format ( "USE %s%n" , finalDbName ) ) ; ddl . add ( String . format ( "ALTER TABLE `%s` CHANGE COLUMN `%s` `%s` %s COMMENT '%s'" , finalTableName , evolvedColumn . getKey ( ) , evolvedColumn . getKey ( ) , evolvedColumn . getValue ( ) , escapeStringForHive ( destinationField . getComment ( ) ) ) ) ; } found = true ; break ; } } if ( ! found ) { String flattenSource = evolvedSchema . getField ( evolvedColumn . getKey ( ) ) . getProp ( "flatten_source" ) ; if ( StringUtils . isBlank ( flattenSource ) ) { flattenSource = evolvedSchema . getField ( evolvedColumn . getKey ( ) ) . name ( ) ; } ddl . add ( String . format ( "USE %s%n" , finalDbName ) ) ; ddl . add ( String . format ( "ALTER TABLE `%s` ADD COLUMNS (`%s` %s COMMENT 'from flatten_source %s')" , finalTableName , evolvedColumn . getKey ( ) , evolvedColumn . getValue ( ) , flattenSource ) ) ; } } return ddl ; }
*
Generate DDLs to evolve final destination table.
@param stagingTableName Staging table.
@param finalTableName Un-evolved final destination table.
@param optionalStagingDbName Optional staging database name, defaults to default.
@param optionalFinalDbName Optional final database name, defaults to default.
@param evolvedSchema Evolved Avro Schema.
@param isEvolutionEnabled Is schema evolution enabled.
@param evolvedColumns Evolved columns in Hive format.
@param destinationTableMeta Destination table metadata.
@return DDLs to evolve final destination table.