001 /* 002 * Copyright (c) 2007-2014 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 }