001    /*
002     * Copyright (c) 2007-2015 Concurrent, Inc. All Rights Reserved.
003     *
004     * Project and contact information: http://www.cascading.org/
005     *
006     * This file is part of the Cascading project.
007     *
008     * Licensed under the Apache License, Version 2.0 (the "License");
009     * you may not use this file except in compliance with the License.
010     * You may obtain a copy of the License at
011     *
012     *     http://www.apache.org/licenses/LICENSE-2.0
013     *
014     * Unless required by applicable law or agreed to in writing, software
015     * distributed under the License is distributed on an "AS IS" BASIS,
016     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
017     * See the License for the specific language governing permissions and
018     * limitations under the License.
019     */
020    
021    package cascading.tap.partition;
022    
023    import java.util.regex.Pattern;
024    
025    import cascading.tuple.Fields;
026    import cascading.tuple.TupleEntry;
027    import cascading.util.Util;
028    
029    /**
030     * DelimitedPartition is an implementation of the {@link Partition} interface that allows for simple
031     * text delimited paths as partitions.
032     * <p/>
033     * For example, given the delimiter {@code -} (dash), a partition path will have dashes.
034     * <p/>
035     * Note the delimiter must not be naturally present in any of the values making up the partition.
036     */
037    public class DelimitedPartition implements Partition
038      {
039      public static final String PATH_DELIM = "/";
040    
041      Fields partitionFields;
042      String delimiter = PATH_DELIM;
043    
044      transient Pattern pattern;
045    
046      public DelimitedPartition( Fields partitionFields, String delimiter )
047        {
048        this( partitionFields );
049    
050        this.delimiter = delimiter;
051        }
052    
053      public DelimitedPartition( Fields partitionFields )
054        {
055        if( partitionFields == null )
056          throw new IllegalArgumentException( "partitionFields must not be null" );
057    
058        if( !partitionFields.isDefined() )
059          throw new IllegalArgumentException( "partitionFields must be defined, got: " + partitionFields.printVerbose() );
060    
061        this.partitionFields = partitionFields;
062        }
063    
064      @Override
065      public int getPathDepth()
066        {
067        return partitionFields.size();
068        }
069    
070      @Override
071      public Fields getPartitionFields()
072        {
073        return partitionFields;
074        }
075    
076      protected Pattern getPattern()
077        {
078        if( pattern == null )
079          pattern = Pattern.compile( delimiter );
080    
081        return pattern;
082        }
083    
084      @Override
085      public void toTuple( String partition, TupleEntry tupleEntry )
086        {
087        if( partition.startsWith( delimiter ) )
088          partition = partition.substring( 1 );
089    
090        String[] split = getPattern().split( partition );
091    
092        tupleEntry.setCanonicalValues( split );
093        }
094    
095      @Override
096      public String toPartition( TupleEntry tupleEntry )
097        {
098        return Util.join( tupleEntry.asIterableOf( String.class ), delimiter, true );
099        }
100      }