001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.hadoop.mapreduce.lib.join;
020
021import java.io.IOException;
022import java.util.ArrayList;
023import java.util.List;
024import java.util.Map;
025import java.util.regex.Matcher;
026import java.util.regex.Pattern;
027
028import org.apache.hadoop.classification.InterfaceAudience;
029import org.apache.hadoop.classification.InterfaceStability;
030import org.apache.hadoop.conf.Configuration;
031import org.apache.hadoop.fs.Path;
032import org.apache.hadoop.io.WritableComparable;
033import org.apache.hadoop.mapreduce.InputFormat;
034import org.apache.hadoop.mapreduce.InputSplit;
035import org.apache.hadoop.mapreduce.JobContext;
036import org.apache.hadoop.mapreduce.RecordReader;
037import org.apache.hadoop.mapreduce.TaskAttemptContext;
038
039/**
040 * An InputFormat capable of performing joins over a set of data sources sorted
041 * and partitioned the same way.
042 *
043 * A user may define new join types by setting the property
044 * <tt>mapreduce.join.define.&lt;ident&gt;</tt> to a classname. 
045 * In the expression <tt>mapreduce.join.expr</tt>, the identifier will be
046 * assumed to be a ComposableRecordReader.
047 * <tt>mapreduce.join.keycomparator</tt> can be a classname used to compare 
048 * keys in the join.
049 * @see #setFormat
050 * @see JoinRecordReader
051 * @see MultiFilterRecordReader
052 */
053@SuppressWarnings("unchecked")
054@InterfaceAudience.Public
055@InterfaceStability.Stable
056public class CompositeInputFormat<K extends WritableComparable>
057    extends InputFormat<K, TupleWritable> {
058
059  public static final String JOIN_EXPR = "mapreduce.join.expr";
060  public static final String JOIN_COMPARATOR = "mapreduce.join.keycomparator";
061  
062  // expression parse tree to which IF requests are proxied
063  private Parser.Node root;
064
065  public CompositeInputFormat() { }
066
067
068  /**
069   * Interpret a given string as a composite expression.
070   * {@code
071   *   func  ::= <ident>([<func>,]*<func>)
072   *   func  ::= tbl(<class>,"<path>")
073   *   class ::= @see java.lang.Class#forName(java.lang.String)
074   *   path  ::= @see org.apache.hadoop.fs.Path#Path(java.lang.String)
075   * }
076   * Reads expression from the <tt>mapreduce.join.expr</tt> property and
077   * user-supplied join types from <tt>mapreduce.join.define.&lt;ident&gt;</tt>
078   *  types. Paths supplied to <tt>tbl</tt> are given as input paths to the
079   * InputFormat class listed.
080   * @see #compose(java.lang.String, java.lang.Class, java.lang.String...)
081   */
082  public void setFormat(Configuration conf) throws IOException {
083    addDefaults();
084    addUserIdentifiers(conf);
085    root = Parser.parse(conf.get(JOIN_EXPR, null), conf);
086  }
087
088  /**
089   * Adds the default set of identifiers to the parser.
090   */
091  protected void addDefaults() {
092    try {
093      Parser.CNode.addIdentifier("inner", InnerJoinRecordReader.class);
094      Parser.CNode.addIdentifier("outer", OuterJoinRecordReader.class);
095      Parser.CNode.addIdentifier("override", OverrideRecordReader.class);
096      Parser.WNode.addIdentifier("tbl", WrappedRecordReader.class);
097    } catch (NoSuchMethodException e) {
098      throw new RuntimeException("FATAL: Failed to init defaults", e);
099    }
100  }
101
102  /**
103   * Inform the parser of user-defined types.
104   */
105  private void addUserIdentifiers(Configuration conf) throws IOException {
106    Pattern x = Pattern.compile("^mapreduce\\.join\\.define\\.(\\w+)$");
107    for (Map.Entry<String,String> kv : conf) {
108      Matcher m = x.matcher(kv.getKey());
109      if (m.matches()) {
110        try {
111          Parser.CNode.addIdentifier(m.group(1),
112              conf.getClass(m.group(0), null, ComposableRecordReader.class));
113        } catch (NoSuchMethodException e) {
114          throw new IOException("Invalid define for " + m.group(1), e);
115        }
116      }
117    }
118  }
119
120  /**
121   * Build a CompositeInputSplit from the child InputFormats by assigning the
122   * ith split from each child to the ith composite split.
123   */
124  @SuppressWarnings("unchecked")
125  public List<InputSplit> getSplits(JobContext job) 
126      throws IOException, InterruptedException {
127    setFormat(job.getConfiguration());
128    job.getConfiguration().setLong("mapreduce.input.fileinputformat.split.minsize", Long.MAX_VALUE);
129    return root.getSplits(job);
130  }
131
132  /**
133   * Construct a CompositeRecordReader for the children of this InputFormat
134   * as defined in the init expression.
135   * The outermost join need only be composable, not necessarily a composite.
136   * Mandating TupleWritable isn't strictly correct.
137   */
138  @SuppressWarnings("unchecked") // child types unknown
139  public RecordReader<K,TupleWritable> createRecordReader(InputSplit split, 
140      TaskAttemptContext taskContext) 
141      throws IOException, InterruptedException {
142    setFormat(taskContext.getConfiguration());
143    return root.createRecordReader(split, taskContext);
144  }
145
146  /**
147   * Convenience method for constructing composite formats.
148   * Given InputFormat class (inf), path (p) return:
149   * {@code tbl(<inf>, <p>) }
150   */
151  public static String compose(Class<? extends InputFormat> inf, 
152      String path) {
153    return compose(inf.getName().intern(), path, 
154             new StringBuffer()).toString();
155  }
156
157  /**
158   * Convenience method for constructing composite formats.
159   * Given operation (op), Object class (inf), set of paths (p) return:
160   * {@code <op>(tbl(<inf>,<p1>),tbl(<inf>,<p2>),...,tbl(<inf>,<pn>)) }
161   */
162  public static String compose(String op, 
163      Class<? extends InputFormat> inf, String... path) {
164    final String infname = inf.getName();
165    StringBuffer ret = new StringBuffer(op + '(');
166    for (String p : path) {
167      compose(infname, p, ret);
168      ret.append(',');
169    }
170    ret.setCharAt(ret.length() - 1, ')');
171    return ret.toString();
172  }
173
174  /**
175   * Convenience method for constructing composite formats.
176   * Given operation (op), Object class (inf), set of paths (p) return:
177   * {@code <op>(tbl(<inf>,<p1>),tbl(<inf>,<p2>),...,tbl(<inf>,<pn>)) }
178   */
179  public static String compose(String op, 
180      Class<? extends InputFormat> inf, Path... path) {
181    ArrayList<String> tmp = new ArrayList<String>(path.length);
182    for (Path p : path) {
183      tmp.add(p.toString());
184    }
185    return compose(op, inf, tmp.toArray(new String[0]));
186  }
187
188  private static StringBuffer compose(String inf, String path,
189      StringBuffer sb) {
190    sb.append("tbl(" + inf + ",\"");
191    sb.append(path);
192    sb.append("\")");
193    return sb;
194  }
195}