coral-beam/src/main/java/com/linkedin/beam/planner/rules/BeamArrayFlattenRule.java
@Override
  public RelNode convert(RelNode rel) {
    final RelDataTypeFactory typeFactory = rel.getCluster().getTypeFactory();
    final LogicalCorrelate correlate = (LogicalCorrelate) rel;
    final RelNode left = getLeft(correlate);
    if (!((left instanceof Project || left instanceof TableScan) && getRight(correlate) instanceof Uncollect)) {
      unsupported();
    }

    final RelDataType leftType = getLeft(correlate).getRowType();
    final List<RelDataType> resultTypes = new ArrayList<>();
    final List<String> resultNames = new ArrayList<>();
    for (RelDataTypeField field : leftType.getFieldList()) {
      resultTypes.add(field.getType());
      resultNames.add(field.getName());
    }

    final Uncollect right = (Uncollect) getRight(correlate);
    if (!(getInput(right) instanceof Project)) {
      unsupported();
    }
    final Project rightInput = (Project) getInput(right);
    final List<String> flattenCols = new ArrayList<>();
    for (RexNode rex : rightInput.getChildExps()) {
      if (rex.getKind() != SqlKind.FIELD_ACCESS) {
        unsupported();
      }
      final RexFieldAccess fieldAccess = (RexFieldAccess) rex;
      final RelDataTypeField leftField = leftType.getField(fieldAccess.getField().getName(), true, false);
      if (fieldAccess.getReferenceExpr().getKind() != SqlKind.CORREL_VARIABLE
          || leftField == null
          || leftField.getType().getComponentType() == null) {
        unsupported();
      }
      flattenCols.add(leftField.getName());
      final RelDataType componentType = leftField.getType().getComponentType();
      if (componentType.isStruct()) {
        for (RelDataTypeField field : componentType.getFieldList()) {
          resultTypes.add(field.getType());
          resultNames.add(field.getName());
        }
      } else {
        resultTypes.add(componentType);
        resultNames.add(correlate.getRowType().getFieldNames().get(resultTypes.size() - 1));
      }
    }

    return BeamArrayFlatten.create(
        convert(correlate.getLeft(), correlate.getLeft().getTraitSet().replace(BeamConvention.INSTANCE)),
        flattenCols,
        typeFactory.createStructType(resultTypes, resultNames));
  }
/**
   * That rule only applies for Pig DataBag flatten, which is translated as correlate operator
   * in Calcite logical plan.
   */
That rule only applies for Pig DataBag flatten, which is translated as correlate operator
in Calcite logical plan.
That rule only applies for Pig DataBag flatten which is translated as correlate operator in Calcite logical plan .
@Override
  public RelNode convert(RelNode rel) {
    final RelDataTypeFactory typeFactory = rel.getCluster().getTypeFactory();
    final LogicalCorrelate correlate = (LogicalCorrelate) rel;
    final RelNode left = getLeft(correlate);
    if (!((left instanceof Project || left instanceof TableScan) && getRight(correlate) instanceof Uncollect)) {
      unsupported();
    }
    final RelDataType leftType = getLeft(correlate).getRowType();
    final List<RelDataType> resultTypes = new ArrayList<>();
    final List<String> resultNames = new ArrayList<>();
    for (RelDataTypeField field : leftType.getFieldList()) {
      resultTypes.add(field.getType());
      resultNames.add(field.getName());
    }
    final Uncollect right = (Uncollect) getRight(correlate);
    if (!(getInput(right) instanceof Project)) {
      unsupported();
    }
    final Project rightInput = (Project) getInput(right);
    final List<String> flattenCols = new ArrayList<>();
    for (RexNode rex : rightInput.getChildExps()) {
      if (rex.getKind() != SqlKind.FIELD_ACCESS) {
        unsupported();
      }
      final RexFieldAccess fieldAccess = (RexFieldAccess) rex;
      final RelDataTypeField leftField = leftType.getField(fieldAccess.getField().getName(), true, false);
      if (fieldAccess.getReferenceExpr().getKind() != SqlKind.CORREL_VARIABLE
          || leftField == null
          || leftField.getType().getComponentType() == null) {
        unsupported();
      }
      flattenCols.add(leftField.getName());
      final RelDataType componentType = leftField.getType().getComponentType();
      if (componentType.isStruct()) {
        for (RelDataTypeField field : componentType.getFieldList()) {
          resultTypes.add(field.getType());
          resultNames.add(field.getName());
        }
      } else {
        resultTypes.add(componentType);
        resultNames.add(correlate.getRowType().getFieldNames().get(resultTypes.size() - 1));
      }
    }
    return BeamArrayFlatten.create(
        convert(correlate.getLeft(), correlate.getLeft().getTraitSet().replace(BeamConvention.INSTANCE)),
        flattenCols,
        typeFactory.createStructType(resultTypes, resultNames));
  }
@ Override public RelNode convert ( RelNode rel ) { final RelDataTypeFactory typeFactory = rel . getCluster ( ) . getTypeFactory ( ) ; final LogicalCorrelate correlate = ( LogicalCorrelate ) rel ; final RelNode left = getLeft ( correlate ) ; if ( ! ( ( left instanceof Project || left instanceof TableScan ) && getRight ( correlate ) instanceof Uncollect ) ) { unsupported ( ) ; } final RelDataType leftType = getLeft ( correlate ) . getRowType ( ) ; final List < RelDataType > resultTypes = new ArrayList < > ( ) ; final List < String > resultNames = new ArrayList < > ( ) ; for ( RelDataTypeField field : leftType . getFieldList ( ) ) { resultTypes . add ( field . getType ( ) ) ; resultNames . add ( field . getName ( ) ) ; } final Uncollect right = ( Uncollect ) getRight ( correlate ) ; if ( ! ( getInput ( right ) instanceof Project ) ) { unsupported ( ) ; } final Project rightInput = ( Project ) getInput ( right ) ; final List < String > flattenCols = new ArrayList < > ( ) ; for ( RexNode rex : rightInput . getChildExps ( ) ) { if ( rex . getKind ( ) != SqlKind . FIELD_ACCESS ) { unsupported ( ) ; } final RexFieldAccess fieldAccess = ( RexFieldAccess ) rex ; final RelDataTypeField leftField = leftType . getField ( fieldAccess . getField ( ) . getName ( ) , true , false ) ; if ( fieldAccess . getReferenceExpr ( ) . getKind ( ) != SqlKind . CORREL_VARIABLE || leftField == null || leftField . getType ( ) . getComponentType ( ) == null ) { unsupported ( ) ; } flattenCols . add ( leftField . getName ( ) ) ; final RelDataType componentType = leftField . getType ( ) . getComponentType ( ) ; if ( componentType . isStruct ( ) ) { for ( RelDataTypeField field : componentType . getFieldList ( ) ) { resultTypes . add ( field . getType ( ) ) ; resultNames . add ( field . getName ( ) ) ; } } else { resultTypes . add ( componentType ) ; resultNames . add ( correlate . getRowType ( ) . getFieldNames ( ) . get ( resultTypes . size ( ) - 1 ) ) ; } } return BeamArrayFlatten . create ( convert ( correlate . getLeft ( ) , correlate . getLeft ( ) . getTraitSet ( ) . replace ( BeamConvention . INSTANCE ) ) , flattenCols , typeFactory . createStructType ( resultTypes , resultNames ) ) ; }
That rule only applies for Pig DataBag flatten, which is translated as correlate operator
 in Calcite logical plan.
That rule only applies for Pig DataBag flatten which is translated as correlate operator in Calcite logical plan .