@ PublicEvolving public TypeComparator < T > createComparator ( int [ ] logicalKeyFields , boolean [ ] orders , int logicalFieldOffset , ExecutionConfig config ) { TypeComparatorBuilder < T > builder = createTypeComparatorBuilder ( ) ; builder . initializeTypeComparatorBuilder ( logicalKeyFields . length ) ; for ( int logicalKeyFieldIndex = 0 ; logicalKeyFieldIndex < logicalKeyFields . length ; logicalKeyFieldIndex ++ ) { int logicalKeyField = logicalKeyFields [ logicalKeyFieldIndex ] ; int logicalField = logicalFieldOffset ; // this is the global/logical field number boolean comparatorAdded = false ; for ( int localFieldId = 0 ; localFieldId < this . getArity ( ) && logicalField <= logicalKeyField && ! comparatorAdded ; localFieldId ++ ) { TypeInformation < ? > localFieldType = this . getTypeAt ( localFieldId ) ; if ( localFieldType instanceof AtomicType && logicalField == logicalKeyField ) { // we found an atomic key --> create comparator builder . addComparatorField ( localFieldId , ( ( AtomicType < ? > ) localFieldType ) . createComparator ( orders [ logicalKeyFieldIndex ] , config ) ) ; comparatorAdded = true ; } // must be composite type and check that the logicalKeyField is within the bounds // of the composite type's logical fields else if ( localFieldType instanceof CompositeType && logicalField <= logicalKeyField && logicalKeyField <= logicalField + ( localFieldType . getTotalFields ( ) - 1 ) ) { // we found a compositeType that is containing the logicalKeyField we are looking for --> create comparator builder . addComparatorField ( localFieldId , ( ( CompositeType < ? > ) localFieldType ) . createComparator ( new int [ ] { logicalKeyField } , new boolean [ ] { orders [ logicalKeyFieldIndex ] } , logicalField , config ) ) ; comparatorAdded = true ; } if ( localFieldType instanceof CompositeType ) { // we need to subtract 1 because we are not accounting for the local field (not accessible for the user) logicalField += localFieldType . getTotalFields ( ) - 1 ; } logicalField ++ ; } if ( ! comparatorAdded ) { throw new IllegalArgumentException ( "Could not add a comparator for the logical" + "key field index " + logicalKeyFieldIndex + "." ) ; } } return builder . createTypeComparator ( config ) ; }
Generic implementation of the comparator creation . Composite types are supplying the infrastructure to create the actual comparators