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.operation; 022 023 import java.io.Serializable; 024 025 import cascading.flow.FlowProcess; 026 import cascading.flow.planner.Scope; 027 import cascading.pipe.Each; 028 import cascading.pipe.Every; 029 import cascading.pipe.Pipe; 030 import cascading.tuple.Fields; 031 import cascading.tuple.Tuple; 032 import cascading.util.Util; 033 034 /** 035 * Class BaseOperation is the base class of predicate types that are applied to {@link Tuple} streams via 036 * the {@link Each} or {@link Every} {@link Pipe}. 037 * </p> 038 * Specific examples of Operations are {@link Function}, {@link Filter}, {@link Aggregator}, {@link Buffer}, 039 * and {@link Assertion}. 040 * <p/> 041 * By default, {@link #isSafe()} returns {@code true}. 042 */ 043 public abstract class BaseOperation<Context> implements Serializable, Operation<Context> 044 { 045 /** Field fieldDeclaration */ 046 protected Fields fieldDeclaration; 047 /** Field numArgs */ 048 protected int numArgs = ANY; 049 /** Field trace */ 050 protected String trace = Util.captureDebugTrace( getClass() ); 051 052 // initialize this operation based on its subclass 053 054 { 055 if( this instanceof Filter || this instanceof Assertion ) 056 fieldDeclaration = Fields.ALL; 057 else 058 fieldDeclaration = Fields.UNKNOWN; 059 } 060 061 /** 062 * Constructs a new instance that returns an {@link Fields#UNKNOWN} {@link Tuple} and accepts any number of arguments. 063 * </p> 064 * It is a best practice to always declare the field names and number of arguments via one of the other constructors. 065 */ 066 protected BaseOperation() 067 { 068 } 069 070 /** 071 * Constructs a new instance that returns the fields declared in fieldDeclaration and accepts any number of arguments. 072 * 073 * @param fieldDeclaration of type Fields 074 */ 075 protected BaseOperation( Fields fieldDeclaration ) 076 { 077 this.fieldDeclaration = fieldDeclaration; 078 verify(); 079 } 080 081 /** 082 * Constructs a new instance that returns an unknown field set and accepts the given numArgs number of arguments. 083 * 084 * @param numArgs of type numArgs 085 */ 086 protected BaseOperation( int numArgs ) 087 { 088 this.numArgs = numArgs; 089 verify(); 090 } 091 092 /** 093 * Constructs a new instance that returns the fields declared in fieldDeclaration and accepts numArgs number of arguments. 094 * 095 * @param numArgs of type numArgs 096 * @param fieldDeclaration of type Fields 097 */ 098 protected BaseOperation( int numArgs, Fields fieldDeclaration ) 099 { 100 this.numArgs = numArgs; 101 this.fieldDeclaration = fieldDeclaration; 102 verify(); 103 } 104 105 /** Validates the state of this instance. */ 106 private final void verify() 107 { 108 if( this instanceof Filter && fieldDeclaration != Fields.ALL ) 109 throw new IllegalArgumentException( "fieldDeclaration must be set to Fields.ALL for filter operations" ); 110 111 if( this instanceof Assertion && fieldDeclaration != Fields.ALL ) 112 throw new IllegalArgumentException( "fieldDeclaration must be set to Fields.ALL for assertion operations" ); 113 114 if( fieldDeclaration == null ) 115 throw new IllegalArgumentException( "fieldDeclaration may not be null" ); 116 117 if( numArgs < 0 ) 118 throw new IllegalArgumentException( "numArgs may not be negative" ); 119 } 120 121 /** Method prepare does nothing, and may safely be overridden. */ 122 @Override 123 public void prepare( FlowProcess flowProcess, OperationCall<Context> operationCall ) 124 { 125 // do nothing 126 } 127 128 @Override 129 public void flush( FlowProcess flowProcess, OperationCall<Context> contextOperationCall ) 130 { 131 } 132 133 /** Method cleanup does nothing, and may safely be overridden. */ 134 @Override 135 public void cleanup( FlowProcess flowProcess, OperationCall<Context> operationCall ) 136 { 137 // do nothing 138 } 139 140 @Override 141 public Fields getFieldDeclaration() 142 { 143 return fieldDeclaration; 144 } 145 146 @Override 147 public int getNumArgs() 148 { 149 return numArgs; 150 } 151 152 @Override 153 public boolean isSafe() 154 { 155 return true; 156 } 157 158 /** 159 * Method getTrace returns the trace of this BaseOperation object. 160 * 161 * @return the trace (type String) of this BaseOperation object. 162 */ 163 public String getTrace() 164 { 165 return trace; 166 } 167 168 @Override 169 public String toString() 170 { 171 return toStringInternal( (Operation) this ); 172 } 173 174 public static String toStringInternal( Operation operation ) 175 { 176 StringBuilder buffer = new StringBuilder(); 177 178 Class<? extends Operation> type = operation.getClass(); 179 if( type.getSimpleName().length() != 0 ) 180 buffer.append( type.getSimpleName() ); 181 else 182 buffer.append( type.getName() ); // should get something for an anonymous inner class 183 184 if( operation.getFieldDeclaration() != null ) 185 buffer.append( "[decl:" ).append( operation.getFieldDeclaration() ).append( "]" ); 186 187 if( operation.getNumArgs() != ANY ) 188 buffer.append( "[args:" ).append( operation.getNumArgs() ).append( "]" ); 189 190 return buffer.toString(); 191 } 192 193 public static void printOperationInternal( Operation operation, StringBuffer buffer, Scope scope ) 194 { 195 Class<? extends Operation> type = operation.getClass(); 196 197 if( type.getSimpleName().length() != 0 ) 198 buffer.append( type.getSimpleName() ); 199 else 200 buffer.append( type.getName() ); // should get something for an anonymous inner class 201 202 if( scope.getOperationDeclaredFields() != null ) 203 buffer.append( "[decl:" ).append( scope.getOperationDeclaredFields().printVerbose() ).append( "]" ); 204 205 if( operation.getNumArgs() != ANY ) 206 buffer.append( "[args:" ).append( operation.getNumArgs() ).append( "]" ); 207 } 208 209 @Override 210 public boolean equals( Object object ) 211 { 212 if( this == object ) 213 return true; 214 if( !( object instanceof BaseOperation ) ) 215 return false; 216 217 BaseOperation that = (BaseOperation) object; 218 219 if( numArgs != that.numArgs ) 220 return false; 221 if( fieldDeclaration != null ? !fieldDeclaration.equals( that.fieldDeclaration ) : that.fieldDeclaration != null ) 222 return false; 223 224 return true; 225 } 226 227 @Override 228 public int hashCode() 229 { 230 int result = fieldDeclaration != null ? fieldDeclaration.hashCode() : 0; 231 result = 31 * result + numArgs; 232 return result; 233 } 234 }