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 final 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 }