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.io; 020 021import java.io.DataInput; 022import java.io.IOException; 023import java.util.concurrent.ConcurrentHashMap; 024 025import org.apache.hadoop.classification.InterfaceAudience; 026import org.apache.hadoop.classification.InterfaceStability; 027import org.apache.hadoop.util.ReflectionUtils; 028 029/** A Comparator for {@link WritableComparable}s. 030 * 031 * <p>This base implemenation uses the natural ordering. To define alternate 032 * orderings, override {@link #compare(WritableComparable,WritableComparable)}. 033 * 034 * <p>One may optimize compare-intensive operations by overriding 035 * {@link #compare(byte[],int,int,byte[],int,int)}. Static utility methods are 036 * provided to assist in optimized implementations of this method. 037 */ 038@InterfaceAudience.Public 039@InterfaceStability.Stable 040public class WritableComparator implements RawComparator { 041 042 private static final ConcurrentHashMap<Class, WritableComparator> comparators 043 = new ConcurrentHashMap<Class, WritableComparator>(); // registry 044 045 /** Get a comparator for a {@link WritableComparable} implementation. */ 046 public static WritableComparator get(Class<? extends WritableComparable> c) { 047 WritableComparator comparator = comparators.get(c); 048 if (comparator == null) { 049 // force the static initializers to run 050 forceInit(c); 051 // look to see if it is defined now 052 comparator = comparators.get(c); 053 // if not, use the generic one 054 if (comparator == null) { 055 comparator = new WritableComparator(c, true); 056 } 057 } 058 return comparator; 059 } 060 061 /** 062 * Force initialization of the static members. 063 * As of Java 5, referencing a class doesn't force it to initialize. Since 064 * this class requires that the classes be initialized to declare their 065 * comparators, we force that initialization to happen. 066 * @param cls the class to initialize 067 */ 068 private static void forceInit(Class<?> cls) { 069 try { 070 Class.forName(cls.getName(), true, cls.getClassLoader()); 071 } catch (ClassNotFoundException e) { 072 throw new IllegalArgumentException("Can't initialize class " + cls, e); 073 } 074 } 075 076 /** Register an optimized comparator for a {@link WritableComparable} 077 * implementation. Comparators registered with this method must be 078 * thread-safe. */ 079 public static void define(Class c, WritableComparator comparator) { 080 comparators.put(c, comparator); 081 } 082 083 private final Class<? extends WritableComparable> keyClass; 084 private final WritableComparable key1; 085 private final WritableComparable key2; 086 private final DataInputBuffer buffer; 087 088 protected WritableComparator() { 089 this(null); 090 } 091 092 /** Construct for a {@link WritableComparable} implementation. */ 093 protected WritableComparator(Class<? extends WritableComparable> keyClass) { 094 this(keyClass, false); 095 } 096 097 protected WritableComparator(Class<? extends WritableComparable> keyClass, 098 boolean createInstances) { 099 this.keyClass = keyClass; 100 if (createInstances) { 101 key1 = newKey(); 102 key2 = newKey(); 103 buffer = new DataInputBuffer(); 104 } else { 105 key1 = key2 = null; 106 buffer = null; 107 } 108 } 109 110 /** Returns the WritableComparable implementation class. */ 111 public Class<? extends WritableComparable> getKeyClass() { return keyClass; } 112 113 /** Construct a new {@link WritableComparable} instance. */ 114 public WritableComparable newKey() { 115 return ReflectionUtils.newInstance(keyClass, null); 116 } 117 118 /** Optimization hook. Override this to make SequenceFile.Sorter's scream. 119 * 120 * <p>The default implementation reads the data into two {@link 121 * WritableComparable}s (using {@link 122 * Writable#readFields(DataInput)}, then calls {@link 123 * #compare(WritableComparable,WritableComparable)}. 124 */ 125 public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { 126 try { 127 buffer.reset(b1, s1, l1); // parse key1 128 key1.readFields(buffer); 129 130 buffer.reset(b2, s2, l2); // parse key2 131 key2.readFields(buffer); 132 133 } catch (IOException e) { 134 throw new RuntimeException(e); 135 } 136 137 return compare(key1, key2); // compare them 138 } 139 140 /** Compare two WritableComparables. 141 * 142 * <p> The default implementation uses the natural ordering, calling {@link 143 * Comparable#compareTo(Object)}. */ 144 @SuppressWarnings("unchecked") 145 public int compare(WritableComparable a, WritableComparable b) { 146 return a.compareTo(b); 147 } 148 149 public int compare(Object a, Object b) { 150 return compare((WritableComparable)a, (WritableComparable)b); 151 } 152 153 /** Lexicographic order of binary data. */ 154 public static int compareBytes(byte[] b1, int s1, int l1, 155 byte[] b2, int s2, int l2) { 156 return FastByteComparisons.compareTo(b1, s1, l1, b2, s2, l2); 157 } 158 159 /** Compute hash for binary data. */ 160 public static int hashBytes(byte[] bytes, int offset, int length) { 161 int hash = 1; 162 for (int i = offset; i < offset + length; i++) 163 hash = (31 * hash) + (int)bytes[i]; 164 return hash; 165 } 166 167 /** Compute hash for binary data. */ 168 public static int hashBytes(byte[] bytes, int length) { 169 return hashBytes(bytes, 0, length); 170 } 171 172 /** Parse an unsigned short from a byte array. */ 173 public static int readUnsignedShort(byte[] bytes, int start) { 174 return (((bytes[start] & 0xff) << 8) + 175 ((bytes[start+1] & 0xff))); 176 } 177 178 /** Parse an integer from a byte array. */ 179 public static int readInt(byte[] bytes, int start) { 180 return (((bytes[start ] & 0xff) << 24) + 181 ((bytes[start+1] & 0xff) << 16) + 182 ((bytes[start+2] & 0xff) << 8) + 183 ((bytes[start+3] & 0xff))); 184 185 } 186 187 /** Parse a float from a byte array. */ 188 public static float readFloat(byte[] bytes, int start) { 189 return Float.intBitsToFloat(readInt(bytes, start)); 190 } 191 192 /** Parse a long from a byte array. */ 193 public static long readLong(byte[] bytes, int start) { 194 return ((long)(readInt(bytes, start)) << 32) + 195 (readInt(bytes, start+4) & 0xFFFFFFFFL); 196 } 197 198 /** Parse a double from a byte array. */ 199 public static double readDouble(byte[] bytes, int start) { 200 return Double.longBitsToDouble(readLong(bytes, start)); 201 } 202 203 /** 204 * Reads a zero-compressed encoded long from a byte array and returns it. 205 * @param bytes byte array with decode long 206 * @param start starting index 207 * @throws java.io.IOException 208 * @return deserialized long 209 */ 210 public static long readVLong(byte[] bytes, int start) throws IOException { 211 int len = bytes[start]; 212 if (len >= -112) { 213 return len; 214 } 215 boolean isNegative = (len < -120); 216 len = isNegative ? -(len + 120) : -(len + 112); 217 if (start+1+len>bytes.length) 218 throw new IOException( 219 "Not enough number of bytes for a zero-compressed integer"); 220 long i = 0; 221 for (int idx = 0; idx < len; idx++) { 222 i = i << 8; 223 i = i | (bytes[start+1+idx] & 0xFF); 224 } 225 return (isNegative ? (i ^ -1L) : i); 226 } 227 228 /** 229 * Reads a zero-compressed encoded integer from a byte array and returns it. 230 * @param bytes byte array with the encoded integer 231 * @param start start index 232 * @throws java.io.IOException 233 * @return deserialized integer 234 */ 235 public static int readVInt(byte[] bytes, int start) throws IOException { 236 return (int) readVLong(bytes, start); 237 } 238}