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
021package cascading.tap.partition;
022
023import java.util.regex.Pattern;
024
025import cascading.tuple.Fields;
026import cascading.tuple.TupleEntry;
027import 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 */
037public 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  }