hive-protolib/src/main/java/com/streamsets/pipeline/stage/lib/hive/HiveMetastoreUtil.java
public static LinkedHashMap<String, HiveTypeInfo> convertRecordToHMSType(
      Record record,
      ELEval scaleEL,
      ELEval precisionEL,
      ELEval commentEL,
      String scaleExpression,
      String precisionExpression,
      String commentExpression,
      ELVars variables
  ) throws HiveStageCheckedException, ELEvalException {
    if(!record.get().getType().isOneOf(Field.Type.MAP, Field.Type.LIST_MAP)) {
      throw new HiveStageCheckedException(Errors.HIVE_33, record.getHeader().getSourceId(), record.get().getType().toString());
    }

    LinkedHashMap<String, HiveTypeInfo> columns = new LinkedHashMap<>();
    Map<String, Field> list = record.get().getValueAsMap();
    for(Map.Entry<String,Field> pair:  list.entrySet()) {
      if (StringUtils.isEmpty(pair.getKey())) {
        throw new HiveStageCheckedException(Errors.HIVE_01, "Field name is empty");
      }
      Field currField = pair.getValue();
      switch(currField.getType()) {
        case SHORT:
          currField = Field.create(Field.Type.INTEGER, currField.getValue());
          break;
        case CHAR:
          currField = Field.create(currField.getValueAsString());
          break;
        case DATETIME:
          currField = Field.create(Field.Type.STRING, currField.getValue() == null ? null : datetimeFormat.get().format(currField.getValueAsDate()));
          break;
        case TIME:
          currField = Field.create(Field.Type.STRING, currField.getValue() == null ? null : timeFormat.get().format(currField.getValueAsTime()));
          break;
        default:
          break;
      }

      // Set current field in the context - used by subsequent ELs (decimal resolution, comments, ...)
      FieldPathEL.setFieldInContext(variables, pair.getKey());

      String comment = commentEL.eval(variables, commentExpression, String.class);
      if(!COMMENT_PATTERN.matcher(comment).matches()) {
        throw new HiveStageCheckedException(com.streamsets.pipeline.stage.processor.hive.Errors.HIVE_METADATA_11, pair.getKey(), comment);
      }

      // Update the Field type and value in Record
      pair.setValue(currField);
      HiveType hiveType = HiveType.getHiveTypeforFieldType(currField.getType());
      HiveTypeInfo hiveTypeInfo;
      // Some types requires special checks or alterations
      if (hiveType == HiveType.DECIMAL) {
        int precision = resolveScaleOrPrecisionExpression("precision", precisionEL, variables, precisionExpression, pair.getKey());
        int scale = resolveScaleOrPrecisionExpression("scale", scaleEL, variables, scaleExpression, pair.getKey());
        validateScaleAndPrecision(pair.getKey(), currField, precision, scale);
        hiveTypeInfo = hiveType.getSupport().generateHiveTypeInfoFromRecordField(currField, comment, precision, scale);
        // We need to make sure that all java objects have the same scale
        if(currField.getValue() != null) {
          pair.setValue(Field.create(currField.getValueAsDecimal().setScale(scale)));
        }
      } else {
        hiveTypeInfo = hiveType.getSupport().generateHiveTypeInfoFromRecordField(currField, comment);
      }
      columns.put(pair.getKey().toLowerCase(), hiveTypeInfo);
    }
    return columns;
  }
public static LinkedHashMap<String, HiveTypeInfo> convertRecordToHMSType(
      Record record,
      ELEval scaleEL,
      ELEval precisionEL,
      ELEval commentEL,
      String scaleExpression,
      String precisionExpression,
      String commentExpression,
      ELVars variables
  ) throws HiveStageCheckedException, ELEvalException {
    if(!record.get().getType().isOneOf(Field.Type.MAP, Field.Type.LIST_MAP)) {
      throw new HiveStageCheckedException(Errors.HIVE_33, record.getHeader().getSourceId(), record.get().getType().toString());
    }

    LinkedHashMap<String, HiveTypeInfo> columns = new LinkedHashMap<>();
    Map<String, Field> list = record.get().getValueAsMap();
    for(Map.Entry<String,Field> pair:  list.entrySet()) {
      if (StringUtils.isEmpty(pair.getKey())) {
        throw new HiveStageCheckedException(Errors.HIVE_01, "Field name is empty");
      }
      Field currField = pair.getValue();
      switch(currField.getType()) {
        case SHORT:
          currField = Field.create(Field.Type.INTEGER, currField.getValue());
          break;
        case CHAR:
          currField = Field.create(currField.getValueAsString());
          break;
        case DATETIME:
          currField = Field.create(Field.Type.STRING, currField.getValue() == null ? null : datetimeFormat.get().format(currField.getValueAsDate()));
          break;
        case TIME:
          currField = Field.create(Field.Type.STRING, currField.getValue() == null ? null : timeFormat.get().format(currField.getValueAsTime()));
          break;
        default:
          break;
      }

      // Set current field in the context - used by subsequent ELs (decimal resolution, comments, ...)
      FieldPathEL.setFieldInContext(variables, pair.getKey());

      String comment = commentEL.eval(variables, commentExpression, String.class);
      if(!COMMENT_PATTERN.matcher(comment).matches()) {
        throw new HiveStageCheckedException(com.streamsets.pipeline.stage.processor.hive.Errors.HIVE_METADATA_11, pair.getKey(), comment);
      }

      // Update the Field type and value in Record
      pair.setValue(currField);
      HiveType hiveType = HiveType.getHiveTypeforFieldType(currField.getType());
      HiveTypeInfo hiveTypeInfo;
      // Some types requires special checks or alterations
      if (hiveType == HiveType.DECIMAL) {
        int precision = resolveScaleOrPrecisionExpression("precision", precisionEL, variables, precisionExpression, pair.getKey());
        int scale = resolveScaleOrPrecisionExpression("scale", scaleEL, variables, scaleExpression, pair.getKey());
        validateScaleAndPrecision(pair.getKey(), currField, precision, scale);
        hiveTypeInfo = hiveType.getSupport().generateHiveTypeInfoFromRecordField(currField, comment, precision, scale);
        // We need to make sure that all java objects have the same scale
        if(currField.getValue() != null) {
          pair.setValue(Field.create(currField.getValueAsDecimal().setScale(scale)));
        }
      } else {
        hiveTypeInfo = hiveType.getSupport().generateHiveTypeInfoFromRecordField(currField, comment);
      }
      columns.put(pair.getKey().toLowerCase(), hiveTypeInfo);
    }
    return columns;
  }
public static LinkedHashMap < String , HiveTypeInfo > convertRecordToHMSType ( Record record , ELEval scaleEL , ELEval precisionEL , ELEval commentEL , String scaleExpression , String precisionExpression , String commentExpression , ELVars variables ) throws HiveStageCheckedException , ELEvalException { if ( ! record . get ( ) . getType ( ) . isOneOf ( Field . Type . MAP , Field . Type . LIST_MAP ) ) { throw new HiveStageCheckedException ( Errors . HIVE_33 , record . getHeader ( ) . getSourceId ( ) , record . get ( ) . getType ( ) . toString ( ) ) ; } LinkedHashMap < String , HiveTypeInfo > columns = new LinkedHashMap <> ( ) ; Map < String , Field > list = record . get ( ) . getValueAsMap ( ) ; for ( Map . Entry < String , Field > pair : list . entrySet ( ) ) { if ( StringUtils . isEmpty ( pair . getKey ( ) ) ) { throw new HiveStageCheckedException ( Errors . HIVE_01 , "Field name is empty" ) ; } Field currField = pair . getValue ( ) ; switch ( currField . getType ( ) ) { case SHORT : currField = Field . create ( Field . Type . INTEGER , currField . getValue ( ) ) ; break ; case CHAR : currField = Field . create ( currField . getValueAsString ( ) ) ; break ; case DATETIME : currField = Field . create ( Field . Type . STRING , currField . getValue ( ) == null ? null : datetimeFormat . get ( ) . format ( currField . getValueAsDate ( ) ) ) ; break ; case TIME : currField = Field . create ( Field . Type . STRING , currField . getValue ( ) == null ? null : timeFormat . get ( ) . format ( currField . getValueAsTime ( ) ) ) ; break ; default : break ; } // Set current field in the context - used by subsequent ELs (decimal resolution, comments, ...) FieldPathEL . setFieldInContext ( variables , pair . getKey ( ) ) ; String comment = commentEL . eval ( variables , commentExpression , String . class ) ; if ( ! COMMENT_PATTERN . matcher ( comment ) . matches ( ) ) { throw new HiveStageCheckedException ( com . streamsets . pipeline . stage . processor . hive . Errors . HIVE_METADATA_11 , pair . getKey ( ) , comment ) ; } // Update the Field type and value in Record pair . setValue ( currField ) ; HiveType hiveType = HiveType . getHiveTypeforFieldType ( currField . getType ( ) ) ; HiveTypeInfo hiveTypeInfo ; // Some types requires special checks or alterations if ( hiveType == HiveType . DECIMAL ) { int precision = resolveScaleOrPrecisionExpression ( "precision" , precisionEL , variables , precisionExpression , pair . getKey ( ) ) ; int scale = resolveScaleOrPrecisionExpression ( "scale" , scaleEL , variables , scaleExpression , pair . getKey ( ) ) ; validateScaleAndPrecision ( pair . getKey ( ) , currField , precision , scale ) ; hiveTypeInfo = hiveType . getSupport ( ) . generateHiveTypeInfoFromRecordField ( currField , comment , precision , scale ) ; // We need to make sure that all java objects have the same scale if ( currField . getValue ( ) != null ) { pair . setValue ( Field . create ( currField . getValueAsDecimal ( ) . setScale ( scale ) ) ) ; } } else { hiveTypeInfo = hiveType . getSupport ( ) . generateHiveTypeInfoFromRecordField ( currField , comment ) ; } columns . put ( pair . getKey ( ) . toLowerCase ( ) , hiveTypeInfo ) ; } return columns ; }
Convert a Record to LinkedHashMap. This is for comparing the structure of incoming Record with cache.
Since Avro does not support char, short, and date types, it needs to convert the type to corresponding
supported types and change the value in record.
@param record incoming Record
@return LinkedHashMap version of record. Key is the column name, and value is column type in HiveType
@throws HiveStageCheckedException
@throws ELEvalException
Convert a Record to LinkedHashMap . This is for comparing the structure of incoming Record with cache . Since Avro does not support char short and date types it needs to convert the type to corresponding supported types and change the value in record .