You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by cj...@apache.org on 2013/09/29 05:30:21 UTC
[1/7] git commit: ACCUMULO-1740 no idea why this fixes the hangs
we've seen
Updated Branches:
refs/heads/ACCUMULO-391 75cccccb3 -> 793b5183b (forced update)
ACCUMULO-1740 no idea why this fixes the hangs we've seen
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/e5e5b2bc
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/e5e5b2bc
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/e5e5b2bc
Branch: refs/heads/ACCUMULO-391
Commit: e5e5b2bc39be0c500e18880d7e1821f2903bfb2d
Parents: acba59b
Author: Eric Newton <er...@gmail.com>
Authored: Fri Sep 27 13:40:08 2013 -0400
Committer: Eric Newton <er...@gmail.com>
Committed: Fri Sep 27 13:40:08 2013 -0400
----------------------------------------------------------------------
.../org/apache/accumulo/server/tabletserver/TabletServer.java | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/e5e5b2bc/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java b/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
index ce22534..575d7e7 100644
--- a/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
+++ b/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
@@ -66,6 +66,8 @@ import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.impl.CompressedIterators;
import org.apache.accumulo.core.client.impl.CompressedIterators.IterConfig;
+import org.apache.accumulo.core.client.impl.Translator.TKeyExtentTranslator;
+import org.apache.accumulo.core.client.impl.Translator.TRangeTranslator;
import org.apache.accumulo.core.client.impl.ScannerImpl;
import org.apache.accumulo.core.client.impl.TabletType;
import org.apache.accumulo.core.client.impl.Translator;
@@ -1325,8 +1327,8 @@ public class TabletServer extends AbstractMetricsImpl implements org.apache.accu
for (ByteBuffer auth : authorizations)
if (!userauths.contains(ByteBufferUtil.toBytes(auth)))
throw new ThriftSecurityException(credentials.getPrincipal(), SecurityErrorCode.BAD_AUTHORIZATIONS);
-
- Map<KeyExtent,List<Range>> batch = Translator.translate(tbatch, Translator.TKET, new Translator.ListTranslator<TRange,Range>(Translator.TRT));
+
+ Map<KeyExtent,List<Range>> batch = Translator.translate(tbatch, new TKeyExtentTranslator(), new Translator.ListTranslator<TRange,Range>(new TRangeTranslator()));
// This is used to determine which thread pool to use
KeyExtent threadPoolExtent = batch.keySet().iterator().next();
[4/7] git commit: The legacy mapred InputFormatBase now verifies (and
fixes the scanner for) a possible change in table name that could happen
between the configuration of the map/reduce job and the actual processing of
the scanner for a specific split.
Posted by cj...@apache.org.
The legacy mapred InputFormatBase now verifies (and fixes the scanner for) a possible change in table name that could happen between the configuration of the map/reduce job and the actual processing of the scanner for a specific split. In that case, the most recent table name associated with the id is always used for the scanner (though the table name that was expected during job setup is still used in the RangeInputSplit). ACCUMULO-391
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/c8a85832
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/c8a85832
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/c8a85832
Branch: refs/heads/ACCUMULO-391
Commit: c8a858323b69d5ba73dc684e9b0033772dbf2119
Parents: fca2731
Author: Corey J. Nolet <cj...@gmail.com>
Authored: Sat Sep 28 23:01:04 2013 -0400
Committer: Corey J. Nolet <cj...@gmail.com>
Committed: Sat Sep 28 23:29:37 2013 -0400
----------------------------------------------------------------------
.../accumulo/core/client/mapred/InputFormatBase.java | 15 ++++++++++++++-
.../client/mapreduce/AccumuloInputFormatTest.java | 1 -
2 files changed, 14 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/c8a85832/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java b/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
index 2d4eadf..fb1b193 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
@@ -594,6 +594,19 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
AuthenticationToken token = getAuthenticationToken(job);
Authorizations authorizations = getScanAuthorizations(job);
+ // in case the table name changed, we can still use the previous name for terms of configuration,
+ // but for the scanner, we'll need to reference the new table name.
+ String actualNameForId = split.getTableName();
+ if(!(instance instanceof MockInstance)) { // Really, the Tables helper class should not be tied to Zookeeper
+ try {
+ actualNameForId = Tables.getTableName(instance, split.getTableId());
+ if (!actualNameForId.equals(split.getTableName()))
+ log.debug("Table name changed from " + split.getTableName() + " to " + actualNameForId);
+ } catch (TableNotFoundException e) {
+ throw new IOException("The specified table was not found for id=" + split.getTableId());
+ }
+ }
+
try {
log.debug("Creating connector with user: " + user);
Connector conn = instance.getConnector(user, token);
@@ -602,7 +615,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
if (isOfflineScan(job)) {
scanner = new OfflineScanner(instance, new Credentials(user, token), Tables.getTableId(instance, getInputTableName(job)), authorizations);
} else {
- scanner = conn.createScanner(getInputTableName(job), authorizations);
+ scanner = conn.createScanner(actualNameForId, authorizations);
}
if (isIsolated(job)) {
log.info("Creating isolated scanner");
http://git-wip-us.apache.org/repos/asf/accumulo/blob/c8a85832/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
index 96a67a6..1b4aa26 100644
--- a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
@@ -331,5 +331,4 @@ public class AccumuloInputFormatTest {
assertEquals(table1, AccumuloInputFormat.getTableQueryConfiguration(job, TEST_TABLE_1));
assertEquals(table2, AccumuloInputFormat.getTableQueryConfiguration(job, TEST_TABLE_2));
}
-
}
[6/7] Adding new TableQueryConfig object for setting multiple table
info in the InputFormatBase
Posted by cj...@apache.org.
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
index c9539c4..0f6655a 100644
--- a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
@@ -16,27 +16,40 @@
*/
package org.apache.accumulo.core.client.mapreduce;
-import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.conf.TableQueryConfig;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.user.RegExFilter;
import org.apache.accumulo.core.iterators.user.WholeRowIterator;
+import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.util.CachedConfiguration;
+import org.apache.accumulo.core.util.Pair;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
@@ -49,11 +62,12 @@ import org.apache.hadoop.util.ToolRunner;
import org.junit.Test;
public class AccumuloInputFormatTest {
-
+
private static final String PREFIX = AccumuloInputFormatTest.class.getSimpleName();
private static final String INSTANCE_NAME = PREFIX + "_mapreduce_instance";
private static final String TEST_TABLE_1 = PREFIX + "_mapreduce_table_1";
-
+ private static final String TEST_TABLE_2 = PREFIX + "_mapreduce_table_2";
+
/**
* Check that the iterator configuration is getting stored in the Job conf correctly.
*
@@ -62,7 +76,7 @@ public class AccumuloInputFormatTest {
@Test
public void testSetIterator() throws IOException {
Job job = new Job();
-
+
IteratorSetting is = new IteratorSetting(1, "WholeRow", "org.apache.accumulo.core.iterators.WholeRowIterator");
AccumuloInputFormat.addIterator(job, is);
Configuration conf = job.getConfiguration();
@@ -71,36 +85,37 @@ public class AccumuloInputFormatTest {
String iterators = conf.get("AccumuloInputFormat.ScanOpts.Iterators");
assertEquals(new String(Base64.encodeBase64(baos.toByteArray())), iterators);
}
-
+
@Test
public void testAddIterator() throws IOException {
Job job = new Job();
-
+
+
AccumuloInputFormat.addIterator(job, new IteratorSetting(1, "WholeRow", WholeRowIterator.class));
AccumuloInputFormat.addIterator(job, new IteratorSetting(2, "Versions", "org.apache.accumulo.core.iterators.VersioningIterator"));
IteratorSetting iter = new IteratorSetting(3, "Count", "org.apache.accumulo.core.iterators.CountingIterator");
iter.addOption("v1", "1");
iter.addOption("junk", "\0omg:!\\xyzzy");
AccumuloInputFormat.addIterator(job, iter);
-
+
List<IteratorSetting> list = AccumuloInputFormat.getIterators(job);
-
+
// Check the list size
assertTrue(list.size() == 3);
-
+
// Walk the list and make sure our settings are correct
IteratorSetting setting = list.get(0);
assertEquals(1, setting.getPriority());
assertEquals("org.apache.accumulo.core.iterators.user.WholeRowIterator", setting.getIteratorClass());
assertEquals("WholeRow", setting.getName());
assertEquals(0, setting.getOptions().size());
-
+
setting = list.get(1);
assertEquals(2, setting.getPriority());
assertEquals("org.apache.accumulo.core.iterators.VersioningIterator", setting.getIteratorClass());
assertEquals("Versions", setting.getName());
assertEquals(0, setting.getOptions().size());
-
+
setting = list.get(2);
assertEquals(3, setting.getPriority());
assertEquals("org.apache.accumulo.core.iterators.CountingIterator", setting.getIteratorClass());
@@ -109,7 +124,7 @@ public class AccumuloInputFormatTest {
assertEquals("1", setting.getOptions().get("v1"));
assertEquals("\0omg:!\\xyzzy", setting.getOptions().get("junk"));
}
-
+
/**
* Test adding iterator options where the keys and values contain both the FIELD_SEPARATOR character (':') and ITERATOR_SEPARATOR (',') characters. There
* should be no exceptions thrown when trying to parse these types of option entries.
@@ -124,12 +139,12 @@ public class AccumuloInputFormatTest {
someSetting.addOption(key, value);
Job job = new Job();
AccumuloInputFormat.addIterator(job, someSetting);
-
+
List<IteratorSetting> list = AccumuloInputFormat.getIterators(job);
assertEquals(1, list.size());
assertEquals(1, list.get(0).getOptions().size());
assertEquals(list.get(0).getOptions().get(key), value);
-
+
someSetting.addOption(key + "2", value);
someSetting.setPriority(2);
someSetting.setName("it2");
@@ -142,7 +157,7 @@ public class AccumuloInputFormatTest {
assertEquals(list.get(1).getOptions().get(key), value);
assertEquals(list.get(1).getOptions().get(key + "2"), value);
}
-
+
/**
* Test getting iterator settings for multiple iterators set
*
@@ -151,69 +166,71 @@ public class AccumuloInputFormatTest {
@Test
public void testGetIteratorSettings() throws IOException {
Job job = new Job();
-
+
AccumuloInputFormat.addIterator(job, new IteratorSetting(1, "WholeRow", "org.apache.accumulo.core.iterators.WholeRowIterator"));
AccumuloInputFormat.addIterator(job, new IteratorSetting(2, "Versions", "org.apache.accumulo.core.iterators.VersioningIterator"));
AccumuloInputFormat.addIterator(job, new IteratorSetting(3, "Count", "org.apache.accumulo.core.iterators.CountingIterator"));
-
+
List<IteratorSetting> list = AccumuloInputFormat.getIterators(job);
-
+
// Check the list size
assertTrue(list.size() == 3);
-
+
// Walk the list and make sure our settings are correct
IteratorSetting setting = list.get(0);
assertEquals(1, setting.getPriority());
assertEquals("org.apache.accumulo.core.iterators.WholeRowIterator", setting.getIteratorClass());
assertEquals("WholeRow", setting.getName());
-
+
setting = list.get(1);
assertEquals(2, setting.getPriority());
assertEquals("org.apache.accumulo.core.iterators.VersioningIterator", setting.getIteratorClass());
assertEquals("Versions", setting.getName());
-
+
setting = list.get(2);
assertEquals(3, setting.getPriority());
assertEquals("org.apache.accumulo.core.iterators.CountingIterator", setting.getIteratorClass());
assertEquals("Count", setting.getName());
-
+
}
-
+
@Test
public void testSetRegex() throws IOException {
Job job = new Job();
-
+
String regex = ">\"*%<>\'\\";
-
+
IteratorSetting is = new IteratorSetting(50, regex, RegExFilter.class);
RegExFilter.setRegexs(is, regex, null, null, null, false);
AccumuloInputFormat.addIterator(job, is);
-
+
assertTrue(regex.equals(AccumuloInputFormat.getIterators(job).get(0).getName()));
}
-
+
private static AssertionError e1 = null;
private static AssertionError e2 = null;
-
+
private static class MRTester extends Configured implements Tool {
+
private static class TestMapper extends Mapper<Key,Value,Key,Value> {
Key key = null;
int count = 0;
-
+
@Override
protected void map(Key k, Value v, Context context) throws IOException, InterruptedException {
try {
+ String tableName = ((InputFormatBase.RangeInputSplit) context.getInputSplit()).getTableName();
if (key != null)
assertEquals(key.getRow().toString(), new String(v.get()));
- assertEquals(k.getRow(), new Text(String.format("%09x", count + 1)));
- assertEquals(new String(v.get()), String.format("%09x", count));
+ assertEquals(new Text(String.format("%s_%09x", tableName, count + 1)), k.getRow());
+ assertEquals(String.format("%s_%09x", tableName, count), new String(v.get()));
} catch (AssertionError e) {
e1 = e;
}
key = new Key(k);
count++;
}
-
+
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
try {
@@ -223,59 +240,198 @@ public class AccumuloInputFormatTest {
}
}
}
-
+
@Override
public int run(String[] args) throws Exception {
-
- if (args.length != 3) {
+
+ if (args.length != 4) {
throw new IllegalArgumentException("Usage : " + MRTester.class.getName() + " <user> <pass> <table>");
}
-
+
String user = args[0];
String pass = args[1];
- String table = args[2];
-
+ String table1 = args[2];
+ String table2 = args[3];
+
Job job = new Job(getConf(), this.getClass().getSimpleName() + "_" + System.currentTimeMillis());
job.setJarByClass(this.getClass());
-
+
job.setInputFormatClass(AccumuloInputFormat.class);
-
+
AccumuloInputFormat.setConnectorInfo(job, user, new PasswordToken(pass));
- AccumuloInputFormat.setInputTableName(job, table);
+
+// AccumuloInputFormat.setInputTableNames(job, Arrays.asList(new String[]{table1, table2}));
+ TableQueryConfig tableConfig1 = new TableQueryConfig(table1);
+ TableQueryConfig tableConfig2 = new TableQueryConfig(table2);
+
+ AccumuloInputFormat.setTableQueryConfigurations(job, tableConfig1, tableConfig2);
AccumuloInputFormat.setMockInstance(job, INSTANCE_NAME);
-
+
job.setMapperClass(TestMapper.class);
job.setMapOutputKeyClass(Key.class);
job.setMapOutputValueClass(Value.class);
job.setOutputFormatClass(NullOutputFormat.class);
-
+
job.setNumReduceTasks(0);
-
+
job.waitForCompletion(true);
-
+
return job.isSuccessful() ? 0 : 1;
}
-
+
public static void main(String[] args) throws Exception {
assertEquals(0, ToolRunner.run(CachedConfiguration.getInstance(), new MRTester(), args));
}
}
-
+
+ /**
+ * Generate incrementing counts and attach table name to the key/value so that order and multi-table data can be verified.
+ */
@Test
public void testMap() throws Exception {
MockInstance mockInstance = new MockInstance(INSTANCE_NAME);
Connector c = mockInstance.getConnector("root", new PasswordToken(""));
c.tableOperations().create(TEST_TABLE_1);
+ c.tableOperations().create(TEST_TABLE_2);
BatchWriter bw = c.createBatchWriter(TEST_TABLE_1, new BatchWriterConfig());
+ BatchWriter bw2 = c.createBatchWriter(TEST_TABLE_2, new BatchWriterConfig());
for (int i = 0; i < 100; i++) {
- Mutation m = new Mutation(new Text(String.format("%09x", i + 1)));
- m.put(new Text(), new Text(), new Value(String.format("%09x", i).getBytes()));
- bw.addMutation(m);
+ Mutation t1m = new Mutation(new Text(String.format("%s_%09x", TEST_TABLE_1, i + 1)));
+ t1m.put(new Text(), new Text(), new Value(String.format("%s_%09x", TEST_TABLE_1, i).getBytes()));
+ bw.addMutation(t1m);
+ Mutation t2m = new Mutation(new Text(String.format("%s_%09x", TEST_TABLE_2, i + 1)));
+ t2m.put(new Text(), new Text(), new Value(String.format("%s_%09x", TEST_TABLE_2, i).getBytes()));
+ bw2.addMutation(t2m);
}
bw.close();
-
- MRTester.main(new String[] {"root", "", TEST_TABLE_1});
+ bw2.close();
+
+ MRTester.main(new String[] {"root", "", TEST_TABLE_1, TEST_TABLE_2});
assertNull(e1);
assertNull(e2);
}
+
+ /**
+ * Asserts that the configuration contains the expected ranges for the tables.
+ */
+ @Test
+ public void testMultitableRangeSerialization() throws Throwable {
+ List<String> tables = Arrays.asList("t1", "t2", "t3");
+ Job job = new Job(new Configuration());
+ job.setInputFormatClass(AccumuloInputFormat.class);
+ job.setMapperClass(MRTester.TestMapper.class);
+ job.setNumReduceTasks(0);
+ AccumuloInputFormat.setConnectorInfo(job, "root", new PasswordToken(new byte[0]));
+ AccumuloInputFormat.setScanAuthorizations(job, new Authorizations());
+ AccumuloInputFormat.setMockInstance(job, "testmapinstance");
+
+ HashMap<String,Collection<Range>> tblRanges = new HashMap<String,Collection<Range>>();
+ for (String tbl : tables) {
+ List<Range> ranges = Arrays.asList(new Range("a", "b"), new Range("c", "d"), new Range("e", "f"));
+ tblRanges.put(tbl, ranges);
+ }
+
+ Range defaultRange = new Range("0", "1");
+
+ try {
+ AccumuloInputFormat.setRanges(job, tblRanges);
+ fail("Exception should have been thrown.");
+ } catch(IllegalStateException e) {}
+
+ AccumuloInputFormat.setInputTableNames(job, tables);
+
+ // set a default range
+ AccumuloInputFormat.setRanges(job, Collections.singleton(defaultRange));
+ AccumuloInputFormat.setRanges(job, tblRanges);
+ Map<String,List<Range>> configuredRanges = AccumuloInputFormat.getRanges(job);
+
+ for (Map.Entry<String,List<Range>> cfgRange : configuredRanges.entrySet()) {
+ String tbl = cfgRange.getKey();
+ HashSet<Range> originalRanges = new HashSet<Range>(tblRanges.remove(tbl));
+ originalRanges.add(defaultRange);
+ HashSet<Range> retrievedRanges = new HashSet<Range>(cfgRange.getValue());
+ assertEquals(originalRanges.size(), retrievedRanges.size());
+ assertTrue(originalRanges.containsAll(retrievedRanges));
+ assertTrue(retrievedRanges.containsAll(originalRanges));
+ }
+ }
+
+ /**
+ * Asserts that the configuration contains the expected iterators for the tables.
+ */
+ @Test
+ public void testMultitableIteratorSerialization() throws Throwable {
+ HashSet<String> tables = new HashSet<String>(Arrays.asList("t1", "t2"));
+ Job job = new Job(new Configuration());
+ job.setInputFormatClass(AccumuloInputFormat.class);
+ job.setMapperClass(MRTester.TestMapper.class);
+ job.setNumReduceTasks(0);
+ AccumuloInputFormat.setConnectorInfo(job, "root", new PasswordToken(new byte[0]));
+ AccumuloInputFormat.setScanAuthorizations(job, new Authorizations());
+
+ // create + set iterators on configuration and build expected reference set
+ IteratorSetting isetting1 = new IteratorSetting(1, "name1", "class1");
+ IteratorSetting isetting2 = new IteratorSetting(2, "name2", "class2");
+ IteratorSetting isetting3 = new IteratorSetting(2, "name3", "class3");
+
+ try {
+ AccumuloInputFormat.addIterator(job, "t1", isetting1);
+ fail("Exception should have been thrown.");
+ } catch(IllegalStateException e) {}
+
+ AccumuloInputFormat.setInputTableNames(job, tables);
+
+ AccumuloInputFormat.addIterator(job, "t1", isetting1);
+ AccumuloInputFormat.addIterator(job, "t2", isetting2);
+ AccumuloInputFormat.addIterator(job, isetting3);
+
+ // verify per-table iterators
+ List<IteratorSetting> t1iters = AccumuloInputFormat.getIterators(job, "t1");
+ List<IteratorSetting> t2iters = AccumuloInputFormat.getIterators(job, "t2");
+ assertFalse(t1iters.isEmpty());
+ assertEquals(isetting1, t1iters.get(1));
+ assertEquals(isetting3, t1iters.get(0));
+ assertEquals(isetting2, t2iters.get(1));
+ assertEquals(isetting3, t2iters.get(0));
+ }
+
+ @Test
+ public void testMultitableColumnSerialization() throws IOException, AccumuloSecurityException {
+ HashSet<String> tables = new HashSet<String>(Arrays.asList("t1", "t2"));
+ Job job = new Job(new Configuration());
+ job.setInputFormatClass(AccumuloInputFormat.class);
+ job.setMapperClass(MRTester.TestMapper.class);
+ job.setNumReduceTasks(0);
+ AccumuloInputFormat.setConnectorInfo(job, "root", new PasswordToken(new byte[0]));
+
+ AccumuloInputFormat.setScanAuthorizations(job, new Authorizations());
+
+ Map<String,Collection<Pair<Text,Text>>> columns = new HashMap<String,Collection<Pair<Text,Text>>>();
+ HashSet<Pair<Text,Text>> t1cols = new HashSet<Pair<Text,Text>>();
+ t1cols.add(new Pair(new Text("a"), new Text("b")));
+ HashSet<Pair<Text,Text>> t2cols = new HashSet<Pair<Text,Text>>();
+ t2cols.add(new Pair(new Text("b"), new Text("c")));
+ columns.put("t1", t1cols);
+ columns.put("t2", t2cols);
+
+ Pair<Text,Text> defaultColumn = new Pair(new Text("c"), new Text("d"));
+
+ try {
+ AccumuloInputFormat.fetchColumns(job, columns);
+ fail("Exception should have been thrown.");
+ } catch(IllegalStateException e) {}
+
+ AccumuloInputFormat.setInputTableNames(job, tables);
+
+ AccumuloInputFormat.fetchColumns(job, Collections.singleton(defaultColumn));
+ AccumuloInputFormat.fetchColumns(job, columns);
+
+ columns.get("t1").add(defaultColumn);
+ columns.get("t2").add(defaultColumn);
+
+ Collection<Pair<Text,Text>> t1actual = AccumuloInputFormat.getFetchedColumns(job, "t1");
+ assertEquals(columns.get("t1"), t1actual);
+ Collection<Pair<Text,Text>> t2actual = AccumuloInputFormat.getFetchedColumns(job, "t2");
+ assertEquals(columns.get("t2"), t2actual);
+ }
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/test/java/org/apache/accumulo/core/conf/TableQueryConfigTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/conf/TableQueryConfigTest.java b/core/src/test/java/org/apache/accumulo/core/conf/TableQueryConfigTest.java
new file mode 100644
index 0000000..1d5d187
--- /dev/null
+++ b/core/src/test/java/org/apache/accumulo/core/conf/TableQueryConfigTest.java
@@ -0,0 +1,93 @@
+package org.apache.accumulo.core.conf;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.util.Pair;
+import org.apache.hadoop.io.Text;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TableQueryConfigTest {
+
+ private static final String TEST_TABLE = "TEST_TABLE";
+ private TableQueryConfig tableQueryConfig;
+
+ @Before
+ public void setUp() {
+ tableQueryConfig = new TableQueryConfig(TEST_TABLE);
+ }
+
+ @Test
+ public void testSerialization_OnlyTable() throws IOException {
+ byte[] serialized = serialize(tableQueryConfig);
+ ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+ TableQueryConfig actualConfig = new TableQueryConfig(new DataInputStream(bais));
+ bais.close();
+
+ assertEquals(tableQueryConfig, actualConfig);
+ }
+
+ @Test
+ public void testSerialization_ranges() throws IOException {
+ List<Range> ranges = new ArrayList<Range>();
+ ranges.add(new Range("a", "b"));
+ ranges.add(new Range("c", "d"));
+ tableQueryConfig.setRanges(ranges);
+
+ byte[] serialized = serialize(tableQueryConfig);
+ ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+ TableQueryConfig actualConfig = new TableQueryConfig(new DataInputStream(bais));
+ bais.close();
+
+ assertEquals(ranges, actualConfig.getRanges());
+ }
+
+ @Test
+ public void testSerialization_columns() throws IOException {
+ Set<Pair<Text,Text>> columns = new HashSet<Pair<Text,Text>>();
+ columns.add(new Pair<Text,Text>(new Text("cf1"), new Text("cq1")));
+ columns.add(new Pair<Text,Text>(new Text("cf2"), null));
+ tableQueryConfig.setColumns(columns);
+
+ byte[] serialized = serialize(tableQueryConfig);
+ ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+ TableQueryConfig actualConfig = new TableQueryConfig(new DataInputStream(bais));
+ bais.close();
+
+ assertEquals(actualConfig.getColumns(), columns);
+ }
+
+ @Test
+ public void testSerialization_iterators() throws IOException {
+ List<IteratorSetting> settings = new ArrayList<IteratorSetting>();
+ settings.add(new IteratorSetting(50, "iter", "iterclass"));
+ settings.add(new IteratorSetting(55, "iter2", "iterclass2"));
+ tableQueryConfig.setIterators(settings);
+ byte[] serialized = serialize(tableQueryConfig);
+ ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+ TableQueryConfig actualConfig = new TableQueryConfig(new DataInputStream(bais));
+ bais.close();
+
+ assertEquals(actualConfig.getIterators(), settings);
+ }
+
+ private byte[] serialize(TableQueryConfig tableQueryConfig) throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ tableQueryConfig.write(new DataOutputStream(baos));
+ baos.close();
+
+ return baos.toByteArray();
+ }
+}
[2/7] git commit: ACCUMULO-1215 stabilize integration tests
Posted by cj...@apache.org.
ACCUMULO-1215 stabilize integration tests
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/9ac95f86
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/9ac95f86
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/9ac95f86
Branch: refs/heads/ACCUMULO-391
Commit: 9ac95f86986a01dfb55ecfdaa1ed3776d2150501
Parents: e5e5b2b
Author: Eric Newton <er...@gmail.com>
Authored: Fri Sep 27 13:41:15 2013 -0400
Committer: Eric Newton <er...@gmail.com>
Committed: Fri Sep 27 13:41:15 2013 -0400
----------------------------------------------------------------------
.../minicluster/MiniAccumuloCluster.java | 17 +++++++++++--
pom.xml | 1 -
.../accumulo/server/util/TServerUtils.java | 2 +-
.../apache/accumulo/test/functional/BulkIT.java | 2 +-
.../accumulo/test/functional/ExamplesIT.java | 26 ++++++++++++--------
.../accumulo/test/functional/LargeRowIT.java | 2 +-
.../accumulo/test/functional/RenameIT.java | 2 +-
.../accumulo/test/functional/RestartIT.java | 4 +--
.../accumulo/test/functional/ShutdownIT.java | 2 +-
.../test/functional/SparseColumnFamilyIT.java | 2 +-
.../accumulo/test/functional/ZooCacheIT.java | 2 +-
test/src/test/resources/log4j.properties | 3 ++-
12 files changed, 42 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
----------------------------------------------------------------------
diff --git a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
index 8805570..77776df 100644
--- a/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
+++ b/minicluster/src/main/java/org/apache/accumulo/minicluster/MiniAccumuloCluster.java
@@ -24,6 +24,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
+import java.net.Socket;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
@@ -264,7 +265,7 @@ public class MiniAccumuloCluster {
File siteFile = new File(config.getConfDir(), "accumulo-site.xml");
writeConfig(siteFile, config.getSiteConfig().entrySet());
-
+
FileWriter fileWriter = new FileWriter(siteFile);
fileWriter.append("<configuration>\n");
@@ -343,7 +344,19 @@ public class MiniAccumuloCluster {
if (!initialized) {
// sleep a little bit to let zookeeper come up before calling init, seems to work better
- UtilWaitThread.sleep(250);
+ while (true) {
+ try {
+ Socket s = new Socket("localhost", config.getZooKeeperPort());
+ s.getOutputStream().write("ruok\n".getBytes());
+ s.getOutputStream().flush();
+ byte buffer[] = new byte[100];
+ int n = s.getInputStream().read(buffer);
+ if (n == 4 && new String(buffer, 0, n).equals("imok"))
+ break;
+ } catch (Exception e) {
+ UtilWaitThread.sleep(250);
+ }
+ }
Process initProcess = exec(Initialize.class, "--instance-name", config.getInstanceName(), "--password", config.getRootPassword());
int ret = initProcess.waitFor();
if (ret != 0) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 62aa809..ae871c0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -108,7 +108,6 @@
</distributionManagement>
<properties>
<accumulo.it.forkCount>1</accumulo.it.forkCount>
- <accumulo.it.threads>1</accumulo.it.threads>
<!-- used for filtering the java source with the current version -->
<accumulo.release.version>${project.version}</accumulo.release.version>
<!-- the maven-release-plugin makes this recommendation, due to plugin bugs -->
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/server/src/main/java/org/apache/accumulo/server/util/TServerUtils.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/accumulo/server/util/TServerUtils.java b/server/src/main/java/org/apache/accumulo/server/util/TServerUtils.java
index fd052c0..1df17fe 100644
--- a/server/src/main/java/org/apache/accumulo/server/util/TServerUtils.java
+++ b/server/src/main/java/org/apache/accumulo/server/util/TServerUtils.java
@@ -298,7 +298,7 @@ public class TServerUtils {
public static ServerAddress startTServer(HostAndPort address, TimedProcessor processor, String serverName, String threadName, int numThreads, long timeBetweenThreadChecks, long maxMessageSize)
throws TTransportException {
ServerAddress result = startHsHaServer(address, processor, serverName, threadName, numThreads, timeBetweenThreadChecks, maxMessageSize);
- // ServerPort result = startThreadPoolServer(port, processor, serverName, threadName, -1);
+ //ServerAddress result = startThreadPoolServer(address, processor, serverName, threadName, -1);
final TServer finalServer = result.server;
Runnable serveTask = new Runnable() {
public void run() {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/java/org/apache/accumulo/test/functional/BulkIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/BulkIT.java b/test/src/test/java/org/apache/accumulo/test/functional/BulkIT.java
index 1ad3fe4..607f2a5 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/BulkIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/BulkIT.java
@@ -30,7 +30,7 @@ public class BulkIT extends SimpleMacIT {
static final int N = 100000;
static final int COUNT = 5;
- @Test(timeout = 2 * 60 * 1000)
+ @Test(timeout = 4 * 60 * 1000)
public void test() throws Exception {
Connector c = getConnector();
String tableName = makeTableName();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/java/org/apache/accumulo/test/functional/ExamplesIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/ExamplesIT.java b/test/src/test/java/org/apache/accumulo/test/functional/ExamplesIT.java
index af2107a..87729e1 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/ExamplesIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/ExamplesIT.java
@@ -169,16 +169,22 @@ public class ExamplesIT extends ConfigurableMacIT {
keepers, "-u", user, "-p", ROOT_PASSWORD, "--num", "100000", "--min", "0", "--max", "1000000000", "--size", "50",
"--batchMemmory", "2M", "--batchLatency", "60s", "--batchThreads", "3", "-t", "bloom_test").waitFor());
c.tableOperations().flush("bloom_test", null, null, true);
- long now = System.currentTimeMillis();
- assertEquals(0, cluster.exec(RandomBatchScanner.class,"--seed", "7", "-i", instance, "-z",
- keepers, "-u", user, "-p", ROOT_PASSWORD, "--num", "10000", "--min", "0", "--max", "1000000000", "--size", "50",
- "--scanThreads", "4","-t", "bloom_test").waitFor());
- long diff = System.currentTimeMillis() - now;
- now = System.currentTimeMillis();
- assertEquals(0, cluster.exec(RandomBatchScanner.class,"--seed", "8", "-i", instance, "-z",
- keepers, "-u", user, "-p", ROOT_PASSWORD, "--num", "10000", "--min", "0", "--max", "1000000000", "--size", "50",
- "--scanThreads", "4","-t", "bloom_test").waitFor());
- long diff2 = System.currentTimeMillis() - now;
+ long diff = 0, diff2 = 0;
+ // try the speed test a couple times in case the system is loaded with other tests
+ for (int i = 0; i < 2; i++) {
+ long now = System.currentTimeMillis();
+ assertEquals(0, cluster.exec(RandomBatchScanner.class,"--seed", "7", "-i", instance, "-z",
+ keepers, "-u", user, "-p", ROOT_PASSWORD, "--num", "10000", "--min", "0", "--max", "1000000000", "--size", "50",
+ "--scanThreads", "4","-t", "bloom_test").waitFor());
+ diff = System.currentTimeMillis() - now;
+ now = System.currentTimeMillis();
+ assertEquals(0, cluster.exec(RandomBatchScanner.class,"--seed", "8", "-i", instance, "-z",
+ keepers, "-u", user, "-p", ROOT_PASSWORD, "--num", "10000", "--min", "0", "--max", "1000000000", "--size", "50",
+ "--scanThreads", "4","-t", "bloom_test").waitFor());
+ diff2 = System.currentTimeMillis() - now;
+ if (diff2 < diff)
+ break;
+ }
assertTrue(diff2 < diff);
log.info("Creating a sharded index of the accumulo java files");
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/java/org/apache/accumulo/test/functional/LargeRowIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/LargeRowIT.java b/test/src/test/java/org/apache/accumulo/test/functional/LargeRowIT.java
index 9493e43..0ef134a 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/LargeRowIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/LargeRowIT.java
@@ -53,7 +53,7 @@ public class LargeRowIT extends ConfigurableMacIT {
private static final int NUM_PRE_SPLITS = 9;
private static final int SPLIT_THRESH = ROW_SIZE * NUM_ROWS / NUM_PRE_SPLITS;
- @Test(timeout = 2 * 60 * 1000)
+ @Test(timeout = 4 * 60 * 1000)
public void run() throws Exception {
Random r = new Random();
byte rowData[] = new byte[ROW_SIZE];
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/java/org/apache/accumulo/test/functional/RenameIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/RenameIT.java b/test/src/test/java/org/apache/accumulo/test/functional/RenameIT.java
index 0758695..ff8d894 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/RenameIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/RenameIT.java
@@ -25,7 +25,7 @@ import org.junit.Test;
public class RenameIT extends SimpleMacIT {
- @Test(timeout = 60 * 1000)
+ @Test(timeout = 2 * 60 * 1000)
public void renameTest() throws Exception {
String name1 = makeTableName();
String name2 = makeTableName();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/java/org/apache/accumulo/test/functional/RestartIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/RestartIT.java b/test/src/test/java/org/apache/accumulo/test/functional/RestartIT.java
index 1567d7b..de29d3b 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/RestartIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/RestartIT.java
@@ -71,7 +71,7 @@ public class RestartIT extends ConfigurableMacIT {
ingest.destroy();
}
- @Test(timeout = 4 * 60 * 1000)
+ @Test(timeout = 8 * 60 * 1000)
public void restartMasterRecovery() throws Exception {
Connector c = getConnector();
c.tableOperations().create("test_ingest");
@@ -108,7 +108,7 @@ public class RestartIT extends ConfigurableMacIT {
ingest.destroy();
}
- @Test(timeout = 2 * 60 * 1000)
+ @Test(timeout = 100 * 60 * 1000)
public void killedTabletServer() throws Exception {
Connector c = getConnector();
c.tableOperations().create("test_ingest");
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/java/org/apache/accumulo/test/functional/ShutdownIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/ShutdownIT.java b/test/src/test/java/org/apache/accumulo/test/functional/ShutdownIT.java
index d60eba6..8d58821 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/ShutdownIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/ShutdownIT.java
@@ -59,7 +59,7 @@ public class ShutdownIT extends ConfigurableMacIT {
}
- @Test(timeout = 60 * 1000)
+ @Test(timeout = 2 * 60 * 1000)
public void shutdownDuringDeleteTable() throws Exception {
final Connector c = getConnector();
for (int i = 0; i < 10 ; i++) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/java/org/apache/accumulo/test/functional/SparseColumnFamilyIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/SparseColumnFamilyIT.java b/test/src/test/java/org/apache/accumulo/test/functional/SparseColumnFamilyIT.java
index 8b7a83a..c45a6a3 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/SparseColumnFamilyIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/SparseColumnFamilyIT.java
@@ -36,7 +36,7 @@ import org.junit.Test;
*/
public class SparseColumnFamilyIT extends SimpleMacIT {
- @Test(timeout = 30 * 1000)
+ @Test(timeout = 60 * 1000)
public void sparceColumnFamily() throws Exception {
String scftt = makeTableName();
Connector c = getConnector();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/java/org/apache/accumulo/test/functional/ZooCacheIT.java
----------------------------------------------------------------------
diff --git a/test/src/test/java/org/apache/accumulo/test/functional/ZooCacheIT.java b/test/src/test/java/org/apache/accumulo/test/functional/ZooCacheIT.java
index 44db530..f4f4964 100644
--- a/test/src/test/java/org/apache/accumulo/test/functional/ZooCacheIT.java
+++ b/test/src/test/java/org/apache/accumulo/test/functional/ZooCacheIT.java
@@ -44,7 +44,7 @@ public class ZooCacheIT extends SimpleMacIT {
reader.start();
threads.add(reader);
}
- assertEquals(0, exec(CacheTestWriter.class, "/zcTest-42", "/tmp/zcTest-42", "3","500").waitFor());
+ assertEquals(0, exec(CacheTestWriter.class, "/zcTest-42", "/tmp/zcTest-42", "3","50").waitFor());
for (Thread t: threads) {
t.join();
if (ref.get() != null)
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9ac95f86/test/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/test/src/test/resources/log4j.properties b/test/src/test/resources/log4j.properties
index fa7ea02..277822a 100644
--- a/test/src/test/resources/log4j.properties
+++ b/test/src/test/resources/log4j.properties
@@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-log4j.rootLogger=INFO, CA
+log4j.rootLogger=DEBUG, CA
log4j.appender.CA=org.apache.log4j.ConsoleAppender
log4j.appender.CA.layout=org.apache.log4j.PatternLayout
log4j.appender.CA.layout.ConversionPattern=%d{ISO8601} [%c{2}] %-5p: %m%n
@@ -29,3 +29,4 @@ log4j.logger.org.apache.hadoop.util.NativeCodeLoader=ERROR
log4j.logger.org.apache.hadoop.util.ProcessTree=WARN
log4j.logger.org.apache.zookeeper.ClientCnxn=FATAL
log4j.logger.org.apache.zookeeper.ZooKeeper=WARN
+log4j.logger.org.apache.accumulo.core.file.rfile.bcfile=INFO
[3/7] git commit: The original single-table setters/getters now
populate a "default" TableQueryConfig object under the hood. This should make
the switch over much easier. Deprecated single table methods in light of the
API changes for the new configurati
Posted by cj...@apache.org.
The original single-table setters/getters now populate a "default" TableQueryConfig object under the hood. This should make the switch over much easier. Deprecated single table methods in light of the API changes for the new configuration object. ACCUMULO-391
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/fca2731f
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/fca2731f
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/fca2731f
Branch: refs/heads/ACCUMULO-391
Commit: fca2731f28827d54514188ac8334ad6f714b12df
Parents: a3f5cdc
Author: Corey J. Nolet <cj...@gmail.com>
Authored: Sat Sep 28 22:53:42 2013 -0400
Committer: Corey J. Nolet <cj...@gmail.com>
Committed: Sat Sep 28 23:29:37 2013 -0400
----------------------------------------------------------------------
.../core/client/mapred/InputFormatBase.java | 17 +-
.../core/client/mapreduce/InputFormatBase.java | 392 ++++++++-----------
.../mapreduce/lib/util/InputConfigurator.java | 147 +++++--
.../accumulo/core/conf/TableQueryConfig.java | 22 +-
.../mapreduce/AccumuloInputFormatTest.java | 128 +-----
5 files changed, 312 insertions(+), 394 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/fca2731f/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java b/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
index ea6bfbc..2d4eadf 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
@@ -276,11 +276,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #setInputTableName(JobConf, String)
*/
protected static String getInputTableName(JobConf job) {
- String[] tableNames = InputConfigurator.getInputTableNames(CLASS, job);
- if (tableNames.length > 0)
- return tableNames[0];
- else
- return null;
+ return InputConfigurator.getInputTableName(CLASS, job);
}
/**
@@ -334,12 +330,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #setRanges(JobConf, Collection)
*/
protected static List<Range> getRanges(JobConf job) throws IOException {
- Map<String, List<Range>> tableRanges = InputConfigurator.getRanges(CLASS,job);
- List<Range> ranges = tableRanges.get(getInputTableName(job));
- if(ranges != null)
- return ranges;
- else
- return new LinkedList<Range>();
+ return InputConfigurator.getRanges(CLASS,job);
}
/**
@@ -366,7 +357,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #fetchColumns(JobConf, Collection)
*/
protected static Set<Pair<Text,Text>> getFetchedColumns(JobConf job) {
- return InputConfigurator.getFetchedColumns(CLASS, job, getInputTableName(job));
+ return InputConfigurator.getFetchedColumns(CLASS, job);
}
/**
@@ -392,7 +383,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #addIterator(JobConf, IteratorSetting)
*/
protected static List<IteratorSetting> getIterators(JobConf job) {
- return InputConfigurator.getDefaultIterators(CLASS,job);
+ return InputConfigurator.getIterators(CLASS,job);
}
/**
http://git-wip-us.apache.org/repos/asf/accumulo/blob/fca2731f/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
index 04ff718..d5208b4 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
@@ -91,10 +91,10 @@ import org.apache.log4j.Logger;
* See {@link AccumuloInputFormat} for an example implementation.
*/
public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
-
+
private static final Class<?> CLASS = AccumuloInputFormat.class;
protected static final Logger log = Logger.getLogger(CLASS);
-
+
/**
* Sets the connector information needed to communicate with Accumulo in this job.
*
@@ -114,7 +114,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void setConnectorInfo(Job job, String principal, AuthenticationToken token) throws AccumuloSecurityException {
InputConfigurator.setConnectorInfo(CLASS, job.getConfiguration(), principal, token);
}
-
+
/**
* Sets the connector information needed to communicate with Accumulo in this job.
*
@@ -133,7 +133,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void setConnectorInfo(Job job, String principal, String tokenFile) throws AccumuloSecurityException {
InputConfigurator.setConnectorInfo(CLASS, job.getConfiguration(), principal, tokenFile);
}
-
+
/**
* Determines if the connector has been configured.
*
@@ -146,7 +146,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static Boolean isConnectorInfoSet(JobContext context) {
return InputConfigurator.isConnectorInfoSet(CLASS, getConfiguration(context));
}
-
+
/**
* Gets the user name from the configuration.
*
@@ -159,7 +159,20 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static String getPrincipal(JobContext context) {
return InputConfigurator.getPrincipal(CLASS, getConfiguration(context));
}
-
+
+ /**
+ * Gets the table name from the configuration.
+ *
+ * @param context
+ * the Hadoop context for the configured job
+ * @return the table name
+ * @since 1.5.0
+ * @see #setInputTableName(Job, String)
+ */
+ protected static String getInputTableName(JobContext context) {
+ return InputConfigurator.getInputTableName(CLASS, getConfiguration(context));
+ }
+
/**
* Gets the serialized token class from either the configuration or the token file.
*
@@ -170,7 +183,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static String getTokenClass(JobContext context) {
return getAuthenticationToken(context).getClass().getName();
}
-
+
/**
* Gets the serialized token from either the configuration or the token file.
*
@@ -181,7 +194,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static byte[] getToken(JobContext context) {
return AuthenticationTokenSerializer.serialize(getAuthenticationToken(context));
}
-
+
/**
* Gets the authenticated token from either the specified token file or directly from the configuration, whichever was used when the job was configured.
*
@@ -193,9 +206,9 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setConnectorInfo(Job, String, String)
*/
protected static AuthenticationToken getAuthenticationToken(JobContext context) {
- return InputConfigurator.getAuthenticationToken(CLASS,getConfiguration(context));
+ return InputConfigurator.getAuthenticationToken(CLASS, getConfiguration(context));
}
-
+
/**
* Configures a {@link ZooKeeperInstance} for this job.
*
@@ -210,7 +223,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void setZooKeeperInstance(Job job, String instanceName, String zooKeepers) {
InputConfigurator.setZooKeeperInstance(CLASS, job.getConfiguration(), instanceName, zooKeepers);
}
-
+
/**
* Configures a {@link MockInstance} for this job.
*
@@ -223,7 +236,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void setMockInstance(Job job, String instanceName) {
InputConfigurator.setMockInstance(CLASS, job.getConfiguration(), instanceName);
}
-
+
/**
* Initializes an Accumulo {@link Instance} based on the configuration.
*
@@ -235,9 +248,9 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setMockInstance(Job, String)
*/
protected static Instance getInstance(JobContext context) {
- return InputConfigurator.getInstance(CLASS,getConfiguration(context));
+ return InputConfigurator.getInstance(CLASS, getConfiguration(context));
}
-
+
/**
* Sets the log level for this job.
*
@@ -248,9 +261,9 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
*/
public static void setLogLevel(Job job, Level level) {
- InputConfigurator.setLogLevel(CLASS,job.getConfiguration(),level);
+ InputConfigurator.setLogLevel(CLASS, job.getConfiguration(), level);
}
-
+
/**
* Gets the log level from this configuration.
*
@@ -263,10 +276,9 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static Level getLogLevel(JobContext context) {
return InputConfigurator.getLogLevel(CLASS, getConfiguration(context));
}
-
+
/**
- * Sets the name of the input table, over which this job will scan. This method has been deprecated in favor of
- * {@link InputFormatBase#setInputTableNames(org.apache.hadoop.mapreduce.Job, java.util.Collection)}
+ * Sets the name of the input table, over which this job will scan.
*
* @param job
* the Hadoop job instance to be configured
@@ -276,26 +288,12 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
*/
@Deprecated
public static void setInputTableName(Job job, String tableName) {
- InputConfigurator.setInputTableName(CLASS,job.getConfiguration(),tableName);
+ InputConfigurator.setInputTableName(CLASS, job.getConfiguration(), tableName);
}
-
- /**
- * Sets the names of the input tables over which this job will scan.
- *
- * @param job
- * the Hadoop job instance to be configured
- * @param tableNames
- * the table to use when the tablename is null in the write call
- * @since 1.6.0
- */
- @Deprecated
- public static void setInputTableNames(Job job, Collection<String> tableNames) {
- InputConfigurator.setInputTableNames(CLASS,job.getConfiguration(),tableNames);
- }
-
+
/**
* Sets the {@link Authorizations} used to scan. Must be a subset of the user's authorization. Defaults to the empty set.
- *
+ *
* @param job
* the Hadoop job instance to be configured
* @param auths
@@ -303,12 +301,12 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
*/
public static void setScanAuthorizations(Job job, Authorizations auths) {
- InputConfigurator.setScanAuthorizations(CLASS,job.getConfiguration(),auths);
+ InputConfigurator.setScanAuthorizations(CLASS, job.getConfiguration(), auths);
}
-
+
/**
* Gets the authorizations to set for the scans from the configuration.
- *
+ *
* @param context
* the Hadoop context for the configured job
* @return the Accumulo scan authorizations
@@ -316,13 +314,13 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setScanAuthorizations(Job, Authorizations)
*/
protected static Authorizations getScanAuthorizations(JobContext context) {
- return InputConfigurator.getScanAuthorizations(CLASS,getConfiguration(context));
+ return InputConfigurator.getScanAuthorizations(CLASS, getConfiguration(context));
}
-
+
/**
* Sets the input ranges to scan for all tables associated with this job. This will be added to any per-table ranges that have been set using
- * {@link #setRanges(org.apache.hadoop.mapreduce.Job, java.util.Map)}
- *
+ * {@link #setRanges(org.apache.hadoop.mapreduce.Job, java.util.Collection)}
+ *
* @param job
* the Hadoop job instance to be configured
* @param ranges
@@ -331,43 +329,26 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
*/
@Deprecated
public static void setRanges(Job job, Collection<Range> ranges) {
- InputConfigurator.setRanges(CLASS,job.getConfiguration(),ranges);
- }
-
- /**
- * Sets the input ranges to scan per-table for this job.
- *
- * @param job
- * the Hadoop job instance to be configured
- * @param ranges
- * the per-table ranges that will be mapped over
- * @since 1.6.0
- */
- @Deprecated
- public static void setRanges(Job job, Map<String,Collection<Range>> ranges) {
InputConfigurator.setRanges(CLASS, job.getConfiguration(), ranges);
}
-
+
/**
* Gets the ranges to scan over from a job.
*
* @param context
* the Hadoop context for the configured job
* @return the ranges
- * @throws IOException
- * if the ranges have been encoded improperly
* @since 1.5.0
* @see #setRanges(Job, Collection)
- * @see #setRanges(org.apache.hadoop.mapreduce.Job, java.util.Map)
+ * @see #setRanges(org.apache.hadoop.mapreduce.Job, java.util.Collection)
*/
@Deprecated
- protected static Map<String,List<Range>> getRanges(JobContext context) throws IOException {
+ protected static List<Range> getRanges(JobContext context) throws IOException {
return InputConfigurator.getRanges(CLASS, getConfiguration(context));
}
-
+
/**
- * Restricts the columns that will be mapped over for this job for all tables. These columns will be added to any per-table columns set with
- * {@link #fetchColumns(org.apache.hadoop.mapreduce.Job, java.util.Map)}.
+ * Restricts the columns that will be mapped over for this job for the default input table.
*
* @param job
* the Hadoop job instance to be configured
@@ -380,38 +361,20 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void fetchColumns(Job job, Collection<Pair<Text,Text>> columnFamilyColumnQualifierPairs) {
InputConfigurator.fetchColumns(CLASS, job.getConfiguration(), columnFamilyColumnQualifierPairs);
}
-
- /**
- * Restricts the columns that will be mapped over for this job per table..
- *
- * @param job
- * the Hadoop job instance to be configured
- * @param columnFamilyColumnQualifierPairs
- * A map keyed by table name where the value is a pair of {@link Text} objects corresponding to column family and column qualifier. If the column
- * qualifier is null, the entire column family is selected. An empty set is the default and is equivalent to scanning the all columns.
- * @since 1.6.0
- */
- @Deprecated
- public static void fetchColumns(Job job, Map<String,Collection<Pair<Text,Text>>> columnFamilyColumnQualifierPairs) {
- InputConfigurator.fetchColumns(CLASS,job.getConfiguration(),columnFamilyColumnQualifierPairs);
- }
-
+
/**
- * Gets the columns to be mapped over from this job. Any default columns as well as per-table columns will be returned.
+ * Gets the columns to be mapped over from this job.
*
* @param context
* the Hadoop context for the configured job
- * @param table
- * the table for which to return the columns
* @return a set of columns
- * @since 1.6.0
+ * @since 1.5.0
* @see #fetchColumns(Job, Collection)
*/
- @Deprecated
- protected static Set<Pair<Text,Text>> getFetchedColumns(JobContext context, String table) {
- return InputConfigurator.getFetchedColumns(CLASS,getConfiguration(context),table);
+ protected static Set<Pair<Text,Text>> getFetchedColumns(JobContext context) {
+ return InputConfigurator.getFetchedColumns(CLASS, getConfiguration(context));
}
-
+
/**
* Encode an iterator on the default all tables for this job.
*
@@ -423,42 +386,9 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
*/
@Deprecated
public static void addIterator(Job job, IteratorSetting cfg) {
- InputConfigurator.addIterator(CLASS,job.getConfiguration(),cfg);
- }
-
- /**
- * Encode an iterator on the input for this job for the specified table.
- *
- * @param job
- * the Hadoop job instance to be configured
- * @param table
- * the table for which to add the iterator
- * @param cfg
- * the configuration of the iterator
- * @since 1.6.0
- */
- @Deprecated
- public static void addIterator(Job job, String table, IteratorSetting cfg) {
- InputConfigurator.addIterator(CLASS, job.getConfiguration(), table, cfg);
+ InputConfigurator.addIterator(CLASS, job.getConfiguration(), cfg);
}
-
- /**
- * Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration for the specific table. Any default iterators will be
- * included in the return.
- *
- * @param context
- * the Hadoop context for the configured job
- * @param table
- * the table for which to return the iterators
- * @return a list of iterators for the given table
- * @since 1.6.0
- * @see #addIterator(Job, String, IteratorSetting)
- */
- @Deprecated
- protected static List<IteratorSetting> getIterators(JobContext context, String table) {
- return InputConfigurator.getIterators(CLASS, getConfiguration(context), table);
- }
-
+
/**
* Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration. This will only return iterators that have not been set
* for a specific table.
@@ -471,9 +401,9 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
*/
@Deprecated
protected static List<IteratorSetting> getIterators(JobContext context) {
- return InputConfigurator.getDefaultIterators(CLASS,getConfiguration(context));
+ return InputConfigurator.getIterators(CLASS, getConfiguration(context));
}
-
+
/**
* Controls the automatic adjustment of ranges for this job. This feature merges overlapping ranges, then splits them to align with tablet boundaries.
* Disabling this feature will cause exactly one Map task to be created for each specified range. The default setting is enabled. *
@@ -492,7 +422,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void setAutoAdjustRanges(Job job, boolean enableFeature) {
InputConfigurator.setAutoAdjustRanges(CLASS, job.getConfiguration(), enableFeature);
}
-
+
/**
* Determines whether a configuration has auto-adjust ranges enabled.
*
@@ -506,21 +436,20 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static boolean getAutoAdjustRanges(JobContext context) {
return InputConfigurator.getAutoAdjustRanges(CLASS, getConfiguration(context));
}
-
-
+
protected static void setTableQueryConfigurations(JobContext job, TableQueryConfig... configs) {
checkNotNull(configs);
- InputConfigurator.setTableQueryConfiguration(CLASS, getConfiguration(job), configs);
+ InputConfigurator.setTableQueryConfigs(CLASS, getConfiguration(job), configs);
}
-
+
public static List<TableQueryConfig> getTableQueryConfigurations(JobContext job) {
- return InputConfigurator.getTableQueryConfigurations(CLASS, getConfiguration(job));
+ return InputConfigurator.getTableQueryConfigs(CLASS, getConfiguration(job));
}
-
+
protected static TableQueryConfig getTableQueryConfiguration(JobContext job, String tableName) {
- return InputConfigurator.getTableQueryConfiguration(CLASS, getConfiguration(job), tableName);
+ return InputConfigurator.getTableQueryConfigs(CLASS, getConfiguration(job), tableName);
}
-
+
/**
* Controls the use of the {@link IsolatedScanner} in this job.
*
@@ -536,7 +465,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void setScanIsolation(Job job, boolean enableFeature) {
InputConfigurator.setScanIsolation(CLASS, job.getConfiguration(), enableFeature);
}
-
+
/**
* Determines whether a configuration has isolation enabled.
*
@@ -549,7 +478,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static boolean isIsolated(JobContext context) {
return InputConfigurator.isIsolated(CLASS, getConfiguration(context));
}
-
+
/**
* Controls the use of the {@link ClientSideIteratorScanner} in this job. Enabling this feature will cause the iterator stack to be constructed within the Map
* task, rather than within the Accumulo TServer. To use this feature, all classes needed for those iterators must be available on the classpath for the task.
@@ -566,7 +495,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void setLocalIterators(Job job, boolean enableFeature) {
InputConfigurator.setLocalIterators(CLASS, job.getConfiguration(), enableFeature);
}
-
+
/**
* Determines whether a configuration uses local iterators.
*
@@ -579,7 +508,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static boolean usesLocalIterators(JobContext context) {
return InputConfigurator.usesLocalIterators(CLASS, getConfiguration(context));
}
-
+
/**
* <p>
* Enable reading offline tables. By default, this feature is disabled and only online tables are scanned. This will make the map reduce job directly read the
@@ -614,7 +543,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static void setOfflineTableScan(Job job, boolean enableFeature) {
InputConfigurator.setOfflineTableScan(CLASS, job.getConfiguration(), enableFeature);
}
-
+
/**
* Determines whether a configuration has the offline table scan feature enabled.
*
@@ -627,7 +556,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static boolean isOfflineScan(JobContext context) {
return InputConfigurator.isOfflineScan(CLASS, getConfiguration(context));
}
-
+
/**
* Initializes an Accumulo {@link TabletLocator} based on the configuration.
*
@@ -643,7 +572,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static TabletLocator getTabletLocator(JobContext context, String table) throws TableNotFoundException {
return InputConfigurator.getTabletLocator(CLASS, getConfiguration(context), table);
}
-
+
// InputFormat doesn't have the equivalent of OutputFormat's checkOutputSpecs(JobContext job)
/**
* Check whether a configuration is fully configured to be used with an Accumulo {@link org.apache.hadoop.mapreduce.InputFormat}.
@@ -657,7 +586,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected static void validateOptions(JobContext context) throws IOException {
InputConfigurator.validateOptions(CLASS, getConfiguration(context));
}
-
+
/**
* An abstract base class to be used to create {@link RecordReader} instances that convert from Accumulo {@link Key}/{@link Value} pairs to the user's K/V
* types.
@@ -674,62 +603,63 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected long numKeysRead;
protected Iterator<Entry<Key,Value>> scannerIterator;
protected RangeInputSplit split;
-
+
/**
- * Apply the configured iterators from the configuration to the scanner. This applies both the
- * default iterators and the per-table iterators.
+ * Apply the configured iterators from the configuration to the scanner. This applies both the default iterators and the per-table iterators.
*
* @param context
* the Hadoop context for the configured job
* @param scanner
* the scanner to configure
- * @param tableName
+ * @param tableName
* the table name for which to set up the iterators
*/
protected void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName) {
- List<IteratorSetting> iterators = getIterators(context, tableName); // default iterators will be included
+ TableQueryConfig config = getTableQueryConfiguration(context, tableName);
+ List<IteratorSetting> iterators = config.getIterators();
for (IteratorSetting iterator : iterators)
scanner.addScanIterator(iterator);
}
-
-
+
/**
* Initialize a scanner over the given input split using this task attempt configuration.
*/
@Override
public void initialize(InputSplit inSplit, TaskAttemptContext attempt) throws IOException {
-
+
Scanner scanner;
split = (RangeInputSplit) inSplit;
log.debug("Initializing input split: " + split.getRange());
Instance instance = getInstance(attempt);
String principal = getPrincipal(attempt);
-
+
TableQueryConfig tableConfig = getTableQueryConfiguration(attempt, split.getTableName());
-
+
// in case the table name changed, we can still use the previous name for terms of configuration,
// but for the scanner, we'll need to reference the new table name.
- String actualNameForId = null;
- try{
- actualNameForId = Tables.getTableName(instance, split.getTableId());
- if(!actualNameForId.equals(split.getTableName()))
- log.debug("Table name changed from " + split.getTableName() + " to " + actualNameForId);
- }catch(TableNotFoundException e){
- throw new IOException("The specified table was not found for id=" + split.getTableId());
+ String actualNameForId = split.getTableName();
+ if(!(instance instanceof MockInstance)) { // Really, the Tables helper class should not be tied to Zookeeper
+ try {
+ actualNameForId = Tables.getTableName(instance, split.getTableId());
+ if (!actualNameForId.equals(split.getTableName()))
+ log.debug("Table name changed from " + split.getTableName() + " to " + actualNameForId);
+ } catch (TableNotFoundException e) {
+ throw new IOException("The specified table was not found for id=" + split.getTableId());
+ }
}
AuthenticationToken token = getAuthenticationToken(attempt);
Authorizations authorizations = getScanAuthorizations(attempt);
try {
log.debug("Creating connector with user: " + principal);
-
+
Connector conn = instance.getConnector(principal, token);
- log.debug("Creating scanner for table: " + actualNameForId);
+ log.debug("Creating scanner for table: " + split.getTableName());
log.debug("Authorizations are: " + authorizations);
if (isOfflineScan(attempt)) {
scanner = new OfflineScanner(instance, new Credentials(principal, token), split.getTableId(), authorizations);
} else {
- scanner = conn.createScanner(actualNameForId, authorizations);
+ scanner = conn.createScanner(split.getTableName(), authorizations);
}
if (isIsolated(attempt)) {
log.info("Creating isolated scanner");
@@ -743,7 +673,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
} catch (Exception e) {
throw new IOException(e);
}
-
+
// setup a scanner within the bounds of this split
for (Pair<Text,Text> c : tableConfig.getColumns()) {
if (c.getSecond() != null) {
@@ -754,64 +684,64 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
scanner.fetchColumnFamily(c.getFirst());
}
}
-
+
scanner.setRange(split.getRange());
numKeysRead = 0;
-
+
// do this last after setting all scanner options
scannerIterator = scanner.iterator();
}
-
+
@Override
public void close() {}
-
+
@Override
public float getProgress() throws IOException {
if (numKeysRead > 0 && currentKey == null)
return 1.0f;
return split.getProgress(currentKey);
}
-
+
protected K currentK = null;
protected V currentV = null;
protected Key currentKey = null;
protected Value currentValue = null;
-
+
@Override
public K getCurrentKey() throws IOException, InterruptedException {
return currentK;
}
-
+
@Override
public V getCurrentValue() throws IOException, InterruptedException {
return currentV;
}
}
-
+
Map<String,Map<KeyExtent,List<Range>>> binOfflineTable(JobContext context, String tableName, List<Range> ranges) throws TableNotFoundException,
AccumuloException, AccumuloSecurityException {
-
+
Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>();
-
+
Instance instance = getInstance(context);
Connector conn = instance.getConnector(getPrincipal(context), getAuthenticationToken(context));
String tableId = Tables.getTableId(instance, tableName);
-
+
if (Tables.getTableState(instance, tableId) != TableState.OFFLINE) {
Tables.clearCache(instance);
if (Tables.getTableState(instance, tableId) != TableState.OFFLINE) {
throw new AccumuloException("Table is online " + tableName + "(" + tableId + ") cannot scan table in offline mode ");
}
}
-
+
for (Range range : ranges) {
Text startRow;
-
+
if (range.getStartKey() != null)
startRow = range.getStartKey().getRow();
else
startRow = new Text();
-
+
Range metadataRange = new Range(new KeyExtent(new Text(tableId), startRow, null).getMetadataEntry(), true, null, false);
Scanner scanner = conn.createScanner(MetadataTable.NAME, Authorizations.EMPTY);
TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.fetch(scanner);
@@ -819,7 +749,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
scanner.fetchColumnFamily(TabletsSection.CurrentLocationColumnFamily.NAME);
scanner.fetchColumnFamily(TabletsSection.FutureLocationColumnFamily.NAME);
scanner.setRange(metadataRange);
-
+
RowIterator rowIter = new RowIterator(scanner);
KeyExtent lastExtent = null;
while (rowIter.hasNext()) {
@@ -827,89 +757,88 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
String last = "";
KeyExtent extent = null;
String location = null;
-
+
while (row.hasNext()) {
Entry<Key,Value> entry = row.next();
Key key = entry.getKey();
-
+
if (key.getColumnFamily().equals(TabletsSection.LastLocationColumnFamily.NAME)) {
last = entry.getValue().toString();
}
-
+
if (key.getColumnFamily().equals(TabletsSection.CurrentLocationColumnFamily.NAME)
|| key.getColumnFamily().equals(TabletsSection.FutureLocationColumnFamily.NAME)) {
location = entry.getValue().toString();
}
-
+
if (TabletsSection.TabletColumnFamily.PREV_ROW_COLUMN.hasColumns(key)) {
extent = new KeyExtent(key.getRow(), entry.getValue());
}
-
+
}
-
+
if (location != null)
return null;
-
+
if (!extent.getTableId().toString().equals(tableId)) {
throw new AccumuloException("Saw unexpected table Id " + tableId + " " + extent);
}
-
+
if (lastExtent != null && !extent.isPreviousExtent(lastExtent)) {
throw new AccumuloException(" " + lastExtent + " is not previous extent " + extent);
}
-
+
Map<KeyExtent,List<Range>> tabletRanges = binnedRanges.get(last);
if (tabletRanges == null) {
tabletRanges = new HashMap<KeyExtent,List<Range>>();
binnedRanges.put(last, tabletRanges);
}
-
+
List<Range> rangeList = tabletRanges.get(extent);
if (rangeList == null) {
rangeList = new ArrayList<Range>();
tabletRanges.put(extent, rangeList);
}
-
+
rangeList.add(range);
-
+
if (extent.getEndRow() == null || range.afterEndKey(new Key(extent.getEndRow()).followingKey(PartialKey.ROW))) {
break;
}
-
+
lastExtent = extent;
}
-
+
}
-
+
return binnedRanges;
}
-
+
/**
* Gets the splits of the tables that have been set on the job.
- *
+ *
* @param conf
* the configuration of the job
- * @return the splits from the tables based on the ranges.
+ * @return the splits from the tables based on the ranges.
* @throws IOException
- * if a table set on the job doesn't exist or an error occurs
- * initializing the tablet locator
+ * if a table set on the job doesn't exist or an error occurs initializing the tablet locator
*/
public List<InputSplit> getSplits(JobContext conf) throws IOException {
log.setLevel(getLogLevel(conf));
validateOptions(conf);
-
+
LinkedList<InputSplit> splits = new LinkedList<InputSplit>();
List<TableQueryConfig> tableConfigs = getTableQueryConfigurations(conf);
for (TableQueryConfig tableConfig : tableConfigs) {
-
- boolean autoAdjust = getAutoAdjustRanges(conf); // TODO: Put this in the table config object
+
+ boolean autoAdjust = tableConfig.shouldAutoAdjustRanges();
String tableId = null;
List<Range> ranges = autoAdjust ? Range.mergeOverlapping(tableConfig.getRanges()) : tableConfig.getRanges();
if (ranges.isEmpty()) {
ranges = new ArrayList<Range>(1);
ranges.add(new Range());
}
-
+
// get the metadata information for these ranges
Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>();
TabletLocator tl;
@@ -920,7 +849,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
// Some tablets were still online, try again
UtilWaitThread.sleep(100 + (int) (Math.random() * 100)); // sleep randomly between 100 and 200 ms
binnedRanges = binOfflineTable(conf, tableConfig.getTableName(), ranges);
-
+
}
} else {
Instance instance = getInstance(conf);
@@ -928,7 +857,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
// its possible that the cache could contain complete, but old information about a tables tablets... so clear it
tl.invalidateCache();
Credentials creds = new Credentials(getPrincipal(conf), getAuthenticationToken(conf));
-
+
while (!tl.binRanges(creds, ranges, binnedRanges).isEmpty()) {
if (!(instance instanceof MockInstance)) {
if (!Tables.exists(instance, tableId))
@@ -946,12 +875,12 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
} catch (Exception e) {
throw new IOException(e);
}
-
+
HashMap<Range,ArrayList<String>> splitsToAdd = null;
-
+
if (!autoAdjust)
splitsToAdd = new HashMap<Range,ArrayList<String>>();
-
+
HashMap<String,String> hostNameCache = new HashMap<String,String>();
for (Entry<String,Map<KeyExtent,List<Range>>> tserverBin : binnedRanges.entrySet()) {
String ip = tserverBin.getKey().split(":", 2)[0];
@@ -978,14 +907,14 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
}
}
}
-
+
if (!autoAdjust)
for (Entry<Range,ArrayList<String>> entry : splitsToAdd.entrySet())
splits.add(new RangeInputSplit(tableConfig.getTableName(), tableId, entry.getKey(), entry.getValue().toArray(new String[0])));
}
return splits;
}
-
+
/**
* The Class RangeInputSplit. Encapsulates an Accumulo range for use in Map Reduce jobs.
*/
@@ -994,51 +923,51 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
private String[] locations;
private String tableId;
private String tableName;
-
+
public RangeInputSplit() {
range = new Range();
locations = new String[0];
tableId = "";
tableName = "";
}
-
+
public RangeInputSplit(RangeInputSplit split) throws IOException {
this.setRange(split.getRange());
this.setLocations(split.getLocations());
this.setTableName(split.getTableName());
}
-
+
protected RangeInputSplit(String table, String tableId, Range range, String[] locations) {
this.range = range;
this.locations = locations;
this.tableName = table;
this.tableId = tableId;
}
-
+
public Range getRange() {
return range;
}
-
+
public void setRange(Range range) {
this.range = range;
}
-
+
public String getTableName() {
return tableName;
}
-
+
public void setTableName(String tableName) {
this.tableName = tableName;
}
-
+
public void setTableId(String tableId) {
this.tableId = tableId;
}
-
+
public String getTableId() {
return tableId;
}
-
+
private static byte[] extractBytes(ByteSequence seq, int numBytes) {
byte[] bytes = new byte[numBytes + 1];
bytes[0] = 0;
@@ -1050,7 +979,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
}
return bytes;
}
-
+
public static float getProgress(ByteSequence start, ByteSequence end, ByteSequence position) {
int maxDepth = Math.min(Math.max(end.length(), start.length()), position.length());
BigInteger startBI = new BigInteger(extractBytes(start, maxDepth));
@@ -1058,7 +987,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
BigInteger positionBI = new BigInteger(extractBytes(position, maxDepth));
return (float) (positionBI.subtract(startBI).doubleValue() / endBI.subtract(startBI).doubleValue());
}
-
+
public float getProgress(Key currentKey) {
if (currentKey == null)
return 0f;
@@ -1077,7 +1006,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
// if we can't figure it out, then claim no progress
return 0f;
}
-
+
/**
* This implementation of length is only an estimate, it does not provide exact values. Do not have your code rely on this return value.
*/
@@ -1087,29 +1016,29 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
Text stopRow = range.isInfiniteStopKey() ? new Text(new byte[] {Byte.MAX_VALUE}) : range.getEndKey().getRow();
int maxCommon = Math.min(7, Math.min(startRow.getLength(), stopRow.getLength()));
long diff = 0;
-
+
byte[] start = startRow.getBytes();
byte[] stop = stopRow.getBytes();
for (int i = 0; i < maxCommon; ++i) {
diff |= 0xff & (start[i] ^ stop[i]);
diff <<= Byte.SIZE;
}
-
+
if (startRow.getLength() != stopRow.getLength())
diff |= 0xff;
-
+
return diff + 1;
}
-
+
@Override
public String[] getLocations() throws IOException {
return locations;
}
-
+
public void setLocations(String[] locations) {
this.locations = locations;
}
-
+
@Override
public void readFields(DataInput in) throws IOException {
range.readFields(in);
@@ -1119,7 +1048,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
for (int i = 0; i < numLocs; ++i)
locations[i] = in.readUTF();
}
-
+
@Override
public void write(DataOutput out) throws IOException {
range.write(out);
@@ -1129,7 +1058,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
out.writeUTF(locations[i]);
}
}
-
+
// use reflection to pull the Configuration out of the JobContext for Hadoop 1 and Hadoop 2 compatibility
static Configuration getConfiguration(JobContext context) {
try {
@@ -1141,5 +1070,4 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
throw new RuntimeException(e);
}
}
-
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/fca2731f/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
index 29795af..5e02177 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
@@ -25,8 +25,10 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashSet;
import java.util.List;
-import java.util.TreeMap;
+import java.util.Set;
+import java.util.StringTokenizer;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
@@ -65,7 +67,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @since 1.5.0
*/
public static enum ScanOpts {
- TABLE, AUTHORIZATIONS, RANGES, COLUMNS, ITERATORS
+ TABLE, AUTHORIZATIONS, RANGES, COLUMNS, ITERATORS, TABLE_CONFIGS
}
/**
@@ -201,6 +203,42 @@ public class InputConfigurator extends ConfiguratorBase {
return ranges;
}
+
+ /**
+ * Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @return a list of iterators
+ * @since 1.5.0
+ * @see #addIterator(Class, Configuration, IteratorSetting)
+ */
+ public static List<IteratorSetting> getIterators(Class<?> implementingClass, Configuration conf) {
+ String iterators = conf.get(enumToConfKey(implementingClass, ScanOpts.ITERATORS));
+
+ // If no iterators are present, return an empty list
+ if (iterators == null || iterators.isEmpty())
+ return new ArrayList<IteratorSetting>();
+
+ // Compose the set of iterators encoded in the job configuration
+ StringTokenizer tokens = new StringTokenizer(iterators, StringUtils.COMMA_STR);
+ List<IteratorSetting> list = new ArrayList<IteratorSetting>();
+ try {
+ while (tokens.hasMoreTokens()) {
+ String itstring = tokens.nextToken();
+ ByteArrayInputStream bais = new ByteArrayInputStream(Base64.decodeBase64(itstring.getBytes()));
+ list.add(new IteratorSetting(new DataInputStream(bais)));
+ bais.close();
+ }
+ } catch (IOException e) {
+ throw new IllegalArgumentException("couldn't decode iterator settings");
+ }
+ return list;
+ }
+
+
/**
* Restricts the columns that will be mapped over for this job. This applies the columns to all tables that have been set on the job.
*
@@ -233,6 +271,29 @@ public class InputConfigurator extends ConfiguratorBase {
conf.setStrings(enumToConfKey(implementingClass, ScanOpts.COLUMNS), columnStrings.toArray(new String[0]));
}
+
+ /**
+ * Gets the columns to be mapped over from this job.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @return a set of columns
+ * @since 1.5.0
+ * @see #fetchColumns(Class, Configuration, Collection)
+ */
+ public static Set<Pair<Text,Text>> getFetchedColumns(Class<?> implementingClass, Configuration conf) {
+ Set<Pair<Text,Text>> columns = new HashSet<Pair<Text,Text>>();
+ for (String col : conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.COLUMNS))) {
+ int idx = col.indexOf(":");
+ Text cf = new Text(idx < 0 ? Base64.decodeBase64(col.getBytes(Constants.UTF8)) : Base64.decodeBase64(col.substring(0, idx).getBytes(Constants.UTF8)));
+ Text cq = idx < 0 ? null : new Text(Base64.decodeBase64(col.substring(idx + 1).getBytes()));
+ columns.add(new Pair<Text,Text>(cf, cq));
+ }
+ return columns;
+ }
+
/**
* Encode an iterator on the input for all tables associated with this job.
*
@@ -269,7 +330,7 @@ public class InputConfigurator extends ConfiguratorBase {
iterators = iterators.concat(StringUtils.COMMA_STR + newIter);
}
// Store the iterators w/ the job
- conf.set(confKey, iterators);
+ conf.set(confKey,iterators);
}
/**
@@ -438,7 +499,7 @@ public class InputConfigurator extends ConfiguratorBase {
return conf.getBoolean(enumToConfKey(implementingClass, Features.SCAN_OFFLINE), false);
}
- public static void setTableQueryConfiguration(Class<?> implementingClass, Configuration conf, TableQueryConfig... tconf) {
+ public static void setTableQueryConfigs(Class<?> implementingClass, Configuration conf, TableQueryConfig... tconf) {
List<String> tableQueryConfigStrings = new ArrayList<String>();
for(TableQueryConfig queryConfig : tconf) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
@@ -449,14 +510,14 @@ public class InputConfigurator extends ConfiguratorBase {
}
tableQueryConfigStrings.add(new String(Base64.encodeBase64(baos.toByteArray())));
}
- String confKey = enumToConfKey(implementingClass, ScanOpts.TABLE);
+ String confKey = enumToConfKey(implementingClass, ScanOpts.TABLE_CONFIGS);
conf.setStrings(confKey, tableQueryConfigStrings.toArray(new String[0]));
}
- public static List<TableQueryConfig> getTableQueryConfigurations(Class<?> implementingClass, Configuration conf) {
+ public static List<TableQueryConfig> getTableQueryConfigs(Class<?> implementingClass, Configuration conf) {
List<TableQueryConfig> configs = new ArrayList<TableQueryConfig>();
- Collection<String> configStrings = conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.TABLE));
- if(configStrings == null) {
+ Collection<String> configStrings = conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.TABLE_CONFIGS));
+ if(configStrings != null) {
for(String str : configStrings) {
try{
byte[] bytes = Base64.decodeBase64(str.getBytes());
@@ -468,24 +529,23 @@ public class InputConfigurator extends ConfiguratorBase {
}
}
}
+ TableQueryConfig defaultQueryConfig;
+ try {
+ defaultQueryConfig = getDefaultTableConfig(implementingClass, conf);
+ } catch(IOException e) {
+ throw new IllegalStateException("There was an error deserializing the default table configuration.");
+ }
+ if(defaultQueryConfig != null)
+ configs.add(defaultQueryConfig);
+
return configs;
}
- public static TableQueryConfig getTableQueryConfiguration(Class<?> implementingClass, Configuration conf, String tableName) {
- Collection<String> configStrings = conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.TABLE));
- if(configStrings == null) {
- for(String str : configStrings) {
- if(str.equals(tableName)) {
- try{
- byte[] bytes = Base64.decodeBase64(str.getBytes());
- ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
- TableQueryConfig config = new TableQueryConfig(new DataInputStream(bais));
- bais.close();
- return config;
- } catch(IOException e) {
- throw new IllegalStateException("The table query configurations could not be deserialized from the given configuration");
- }
- }
+ public static TableQueryConfig getTableQueryConfigs(Class<?> implementingClass, Configuration conf, String tableName) {
+ List<TableQueryConfig> queryConfigs = getTableQueryConfigs(implementingClass,conf);
+ for(TableQueryConfig queryConfig : queryConfigs) {
+ if(queryConfig.getTableName().equals(tableName)) {
+ return queryConfig;
}
}
return null;
@@ -509,7 +569,7 @@ public class InputConfigurator extends ConfiguratorBase {
String instanceType = conf.get(enumToConfKey(implementingClass, InstanceOpts.TYPE));
if ("MockInstance".equals(instanceType))
return new MockTabletLocator();
- Instance instance = getInstance(implementingClass, conf);
+ Instance instance = getInstance(implementingClass,conf);
return TabletLocator.getLocator(instance, new Text(Tables.getTableId(instance, tableName)));
}
@@ -539,19 +599,19 @@ public class InputConfigurator extends ConfiguratorBase {
if (!c.securityOperations().authenticateUser(principal, token))
throw new IOException("Unable to authenticate user");
- for (TableQueryConfig tableConfig : getTableQueryConfigurations(implementingClass, conf)) {
+ for (TableQueryConfig tableConfig : getTableQueryConfigs(implementingClass,conf)) {
if (!c.securityOperations().hasTablePermission(getPrincipal(implementingClass, conf), tableConfig.getTableName(), TablePermission.READ))
throw new IOException("Unable to access table");
}
- for (TableQueryConfig tableConfig : getTableQueryConfigurations(implementingClass,conf)) {
+ for (TableQueryConfig tableConfig : getTableQueryConfigs(implementingClass,conf)) {
if(!tableConfig.shouldUseLocalIterators()) {
-// if (!conf.getBoolean(enumToConfKey(implementingClass, Features.USE_LOCAL_ITERATORS), false)) {
- // validate that any scan-time iterators can be loaded by the the tablet servers
- for (IteratorSetting iter : tableConfig.getIterators()) { // TODO: These iterators need to be separated by table
- if (!c.tableOperations().testClassLoad(tableConfig.getTableName(), iter.getIteratorClass(), SortedKeyValueIterator.class.getName()))
- throw new AccumuloException("Servers are unable to load " + iter.getIteratorClass() + " as a " + SortedKeyValueIterator.class.getName());
+ if(tableConfig.getIterators() != null) {
+ for (IteratorSetting iter : tableConfig.getIterators()) {
+ if (!c.tableOperations().testClassLoad(tableConfig.getTableName(), iter.getIteratorClass(), SortedKeyValueIterator.class.getName()))
+ throw new AccumuloException("Servers are unable to load " + iter.getIteratorClass() + " as a " + SortedKeyValueIterator.class.getName());
+ }
}
}
}
@@ -567,4 +627,29 @@ public class InputConfigurator extends ConfiguratorBase {
}
}
+
+ protected static TableQueryConfig getDefaultTableConfig(Class<?> implementingClass, Configuration conf) throws IOException {
+ String tableName = getInputTableName(implementingClass, conf);
+ if(tableName != null) {
+ TableQueryConfig queryConfig = new TableQueryConfig(getInputTableName(implementingClass, conf));
+ List<IteratorSetting> itrs = getIterators(implementingClass, conf);
+ if(itrs != null)
+ queryConfig.setIterators(itrs);
+ Set<Pair<Text,Text>> columns = getFetchedColumns(implementingClass, conf);
+ if(columns != null)
+ queryConfig.setColumns(columns);
+ List<Range> ranges = getRanges(implementingClass, conf);
+ if(ranges != null)
+ queryConfig.setRanges(ranges);
+
+ queryConfig.setAutoAdjustRanges(getAutoAdjustRanges(implementingClass, conf))
+ .setUseIsolatedScanners(isIsolated(implementingClass, conf))
+ .setUseLocalIterators(usesLocalIterators(implementingClass, conf))
+ .setOfflineScan(isOfflineScan(implementingClass, conf));
+ return queryConfig;
+ }
+ return null;
+ }
+
+
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/fca2731f/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java b/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java
index 00f4dc0..d6476da 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java
@@ -27,6 +27,7 @@ public class TableQueryConfig implements Writable {
private boolean autoAdjustRanges = true;
private boolean useLocalIterators = false;
private boolean useIsolatedScanners = false;
+ private boolean offlineScan = false;
public TableQueryConfig(String tableName) {
checkNotNull(tableName);
@@ -67,20 +68,29 @@ public class TableQueryConfig implements Writable {
return this;
}
+ public boolean isOfflineScan(){
+ return offlineScan;
+ }
+
+ public TableQueryConfig setOfflineScan(boolean offlineScan){
+ this.offlineScan=offlineScan;
+ return this;
+ }
+
public String getTableName(){
return tableName;
}
public List<IteratorSetting> getIterators(){
- return iterators;
+ return iterators != null ? iterators : new ArrayList<IteratorSetting>();
}
public List<Range> getRanges(){
- return ranges;
+ return ranges != null ? ranges : new ArrayList<Range>();
}
public Set<Pair<Text,Text>> getColumns(){
- return columns;
+ return columns != null ? columns : new HashSet<Pair<Text,Text>>();
}
public boolean shouldAutoAdjustRanges(){
@@ -179,6 +189,9 @@ public class TableQueryConfig implements Writable {
TableQueryConfig that=(TableQueryConfig)o;
if(autoAdjustRanges!=that.autoAdjustRanges) return false;
+ if(offlineScan!=that.offlineScan) return false;
+ if(useIsolatedScanners!=that.useIsolatedScanners) return false;
+ if(useLocalIterators!=that.useLocalIterators) return false;
if(columns!=null?!columns.equals(that.columns):that.columns!=null) return false;
if(iterators!=null?!iterators.equals(that.iterators):that.iterators!=null) return false;
if(ranges!=null?!ranges.equals(that.ranges):that.ranges!=null) return false;
@@ -194,6 +207,9 @@ public class TableQueryConfig implements Writable {
result=31*result+(ranges!=null?ranges.hashCode():0);
result=31*result+(columns!=null?columns.hashCode():0);
result=31*result+(autoAdjustRanges?1:0);
+ result=31*result+(useLocalIterators?1:0);
+ result=31*result+(useIsolatedScanners?1:0);
+ result=31*result+(offlineScan?1:0);
return result;
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/fca2731f/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
index 0f6655a..96a67a6 100644
--- a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
@@ -311,127 +311,25 @@ public class AccumuloInputFormatTest {
assertNull(e2);
}
- /**
- * Asserts that the configuration contains the expected ranges for the tables.
- */
@Test
- public void testMultitableRangeSerialization() throws Throwable {
- List<String> tables = Arrays.asList("t1", "t2", "t3");
- Job job = new Job(new Configuration());
- job.setInputFormatClass(AccumuloInputFormat.class);
- job.setMapperClass(MRTester.TestMapper.class);
- job.setNumReduceTasks(0);
- AccumuloInputFormat.setConnectorInfo(job, "root", new PasswordToken(new byte[0]));
- AccumuloInputFormat.setScanAuthorizations(job, new Authorizations());
- AccumuloInputFormat.setMockInstance(job, "testmapinstance");
-
- HashMap<String,Collection<Range>> tblRanges = new HashMap<String,Collection<Range>>();
- for (String tbl : tables) {
- List<Range> ranges = Arrays.asList(new Range("a", "b"), new Range("c", "d"), new Range("e", "f"));
- tblRanges.put(tbl, ranges);
- }
+ public void testTableQueryConfigSerialization() throws IOException{
- Range defaultRange = new Range("0", "1");
-
- try {
- AccumuloInputFormat.setRanges(job, tblRanges);
- fail("Exception should have been thrown.");
- } catch(IllegalStateException e) {}
+ Job job = new Job();
- AccumuloInputFormat.setInputTableNames(job, tables);
+ TableQueryConfig table1 = new TableQueryConfig(TEST_TABLE_1)
+ .setRanges(Collections.singletonList(new Range("a", "b")))
+ .setColumns(Collections.singleton(new Pair<Text,Text>(new Text("CF1"), new Text("CQ1"))))
+ .setIterators(Collections.singletonList(new IteratorSetting(50, "iter1", "iterclass1")));
- // set a default range
- AccumuloInputFormat.setRanges(job, Collections.singleton(defaultRange));
- AccumuloInputFormat.setRanges(job, tblRanges);
- Map<String,List<Range>> configuredRanges = AccumuloInputFormat.getRanges(job);
+ TableQueryConfig table2 = new TableQueryConfig(TEST_TABLE_2)
+ .setRanges(Collections.singletonList(new Range("a", "b")))
+ .setColumns(Collections.singleton(new Pair<Text,Text>(new Text("CF1"), new Text("CQ1"))))
+ .setIterators(Collections.singletonList(new IteratorSetting(50, "iter1", "iterclass1")));
- for (Map.Entry<String,List<Range>> cfgRange : configuredRanges.entrySet()) {
- String tbl = cfgRange.getKey();
- HashSet<Range> originalRanges = new HashSet<Range>(tblRanges.remove(tbl));
- originalRanges.add(defaultRange);
- HashSet<Range> retrievedRanges = new HashSet<Range>(cfgRange.getValue());
- assertEquals(originalRanges.size(), retrievedRanges.size());
- assertTrue(originalRanges.containsAll(retrievedRanges));
- assertTrue(retrievedRanges.containsAll(originalRanges));
- }
- }
+ AccumuloInputFormat.setTableQueryConfigurations(job, table1, table2);
- /**
- * Asserts that the configuration contains the expected iterators for the tables.
- */
- @Test
- public void testMultitableIteratorSerialization() throws Throwable {
- HashSet<String> tables = new HashSet<String>(Arrays.asList("t1", "t2"));
- Job job = new Job(new Configuration());
- job.setInputFormatClass(AccumuloInputFormat.class);
- job.setMapperClass(MRTester.TestMapper.class);
- job.setNumReduceTasks(0);
- AccumuloInputFormat.setConnectorInfo(job, "root", new PasswordToken(new byte[0]));
- AccumuloInputFormat.setScanAuthorizations(job, new Authorizations());
-
- // create + set iterators on configuration and build expected reference set
- IteratorSetting isetting1 = new IteratorSetting(1, "name1", "class1");
- IteratorSetting isetting2 = new IteratorSetting(2, "name2", "class2");
- IteratorSetting isetting3 = new IteratorSetting(2, "name3", "class3");
-
- try {
- AccumuloInputFormat.addIterator(job, "t1", isetting1);
- fail("Exception should have been thrown.");
- } catch(IllegalStateException e) {}
-
- AccumuloInputFormat.setInputTableNames(job, tables);
-
- AccumuloInputFormat.addIterator(job, "t1", isetting1);
- AccumuloInputFormat.addIterator(job, "t2", isetting2);
- AccumuloInputFormat.addIterator(job, isetting3);
-
- // verify per-table iterators
- List<IteratorSetting> t1iters = AccumuloInputFormat.getIterators(job, "t1");
- List<IteratorSetting> t2iters = AccumuloInputFormat.getIterators(job, "t2");
- assertFalse(t1iters.isEmpty());
- assertEquals(isetting1, t1iters.get(1));
- assertEquals(isetting3, t1iters.get(0));
- assertEquals(isetting2, t2iters.get(1));
- assertEquals(isetting3, t2iters.get(0));
+ assertEquals(table1, AccumuloInputFormat.getTableQueryConfiguration(job, TEST_TABLE_1));
+ assertEquals(table2, AccumuloInputFormat.getTableQueryConfiguration(job, TEST_TABLE_2));
}
- @Test
- public void testMultitableColumnSerialization() throws IOException, AccumuloSecurityException {
- HashSet<String> tables = new HashSet<String>(Arrays.asList("t1", "t2"));
- Job job = new Job(new Configuration());
- job.setInputFormatClass(AccumuloInputFormat.class);
- job.setMapperClass(MRTester.TestMapper.class);
- job.setNumReduceTasks(0);
- AccumuloInputFormat.setConnectorInfo(job, "root", new PasswordToken(new byte[0]));
-
- AccumuloInputFormat.setScanAuthorizations(job, new Authorizations());
-
- Map<String,Collection<Pair<Text,Text>>> columns = new HashMap<String,Collection<Pair<Text,Text>>>();
- HashSet<Pair<Text,Text>> t1cols = new HashSet<Pair<Text,Text>>();
- t1cols.add(new Pair(new Text("a"), new Text("b")));
- HashSet<Pair<Text,Text>> t2cols = new HashSet<Pair<Text,Text>>();
- t2cols.add(new Pair(new Text("b"), new Text("c")));
- columns.put("t1", t1cols);
- columns.put("t2", t2cols);
-
- Pair<Text,Text> defaultColumn = new Pair(new Text("c"), new Text("d"));
-
- try {
- AccumuloInputFormat.fetchColumns(job, columns);
- fail("Exception should have been thrown.");
- } catch(IllegalStateException e) {}
-
- AccumuloInputFormat.setInputTableNames(job, tables);
-
- AccumuloInputFormat.fetchColumns(job, Collections.singleton(defaultColumn));
- AccumuloInputFormat.fetchColumns(job, columns);
-
- columns.get("t1").add(defaultColumn);
- columns.get("t2").add(defaultColumn);
-
- Collection<Pair<Text,Text>> t1actual = AccumuloInputFormat.getFetchedColumns(job, "t1");
- assertEquals(columns.get("t1"), t1actual);
- Collection<Pair<Text,Text>> t2actual = AccumuloInputFormat.getFetchedColumns(job, "t2");
- assertEquals(columns.get("t2"), t2actual);
- }
}
[7/7] git commit: Adding new TableQueryConfig object for setting
multiple table info in the InputFormatBase
Posted by cj...@apache.org.
Adding new TableQueryConfig object for setting multiple table info in the InputFormatBase
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/a3f5cdc6
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/a3f5cdc6
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/a3f5cdc6
Branch: refs/heads/ACCUMULO-391
Commit: a3f5cdc6b9b0d602efba11b87361ca90bc551fa1
Parents: 9ac95f8
Author: Corey J. Nolet <cj...@gmail.com>
Authored: Sat Sep 28 21:58:40 2013 -0400
Committer: Corey J. Nolet <cj...@gmail.com>
Committed: Sat Sep 28 23:29:37 2013 -0400
----------------------------------------------------------------------
.../core/client/impl/thrift/ClientService.java | 16 +-
.../thrift/ThriftTableOperationException.java | 4 +-
.../core/client/mapred/InputFormatBase.java | 51 ++-
.../core/client/mapreduce/InputFormatBase.java | 397 +++++++++++++------
.../mapreduce/lib/util/InputConfigurator.java | 226 ++++++-----
.../accumulo/core/conf/TableQueryConfig.java | 199 ++++++++++
.../core/master/thrift/MasterClientService.java | 16 +-
.../accumulo/core/util/ArgumentChecker.java | 5 +
.../client/mapred/AccumuloInputFormatTest.java | 10 +-
.../mapreduce/AccumuloInputFormatTest.java | 260 +++++++++---
.../core/conf/TableQueryConfigTest.java | 93 +++++
11 files changed, 965 insertions(+), 312 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ClientService.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ClientService.java b/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ClientService.java
index 488e065..f44d4a6 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ClientService.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ClientService.java
@@ -17751,7 +17751,7 @@ import org.slf4j.LoggerFactory;
return CREDENTIALS;
case 2: // PRINCIPAL
return PRINCIPAL;
- case 3: // TABLE_NAME
+ case 3: // TABLE
return TABLE_NAME;
case 4: // TBL_PERM
return TBL_PERM;
@@ -18329,7 +18329,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 3: // TABLE_NAME
+ case 3: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
@@ -21107,7 +21107,7 @@ import org.slf4j.LoggerFactory;
return CREDENTIALS;
case 2: // PRINCIPAL
return PRINCIPAL;
- case 3: // TABLE_NAME
+ case 3: // TABLE
return TABLE_NAME;
case 4: // PERMISSION
return PERMISSION;
@@ -21685,7 +21685,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 3: // TABLE_NAME
+ case 3: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
@@ -22327,7 +22327,7 @@ import org.slf4j.LoggerFactory;
return CREDENTIALS;
case 2: // PRINCIPAL
return PRINCIPAL;
- case 3: // TABLE_NAME
+ case 3: // TABLE
return TABLE_NAME;
case 4: // PERMISSION
return PERMISSION;
@@ -22905,7 +22905,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 3: // TABLE_NAME
+ case 3: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
@@ -24536,7 +24536,7 @@ import org.slf4j.LoggerFactory;
return TINFO;
case 3: // CREDENTIALS
return CREDENTIALS;
- case 2: // TABLE_NAME
+ case 2: // TABLE
return TABLE_NAME;
default:
return null;
@@ -24960,7 +24960,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 2: // TABLE_NAME
+ case 2: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ThriftTableOperationException.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ThriftTableOperationException.java b/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ThriftTableOperationException.java
index 3863e75..dd2e192 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ThriftTableOperationException.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/thrift/ThriftTableOperationException.java
@@ -106,7 +106,7 @@ import org.slf4j.LoggerFactory;
switch(fieldId) {
case 1: // TABLE_ID
return TABLE_ID;
- case 2: // TABLE_NAME
+ case 2: // TABLE
return TABLE_NAME;
case 3: // OP
return OP;
@@ -680,7 +680,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 2: // TABLE_NAME
+ case 2: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java b/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
index c796cd2..ea6bfbc 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -132,7 +133,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #setConnectorInfo(JobConf, String, AuthenticationToken)
*/
protected static Boolean isConnectorInfoSet(JobConf job) {
- return InputConfigurator.isConnectorInfoSet(CLASS, job);
+ return InputConfigurator.isConnectorInfoSet(CLASS,job);
}
/**
@@ -275,7 +276,11 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #setInputTableName(JobConf, String)
*/
protected static String getInputTableName(JobConf job) {
- return InputConfigurator.getInputTableName(CLASS, job);
+ String[] tableNames = InputConfigurator.getInputTableNames(CLASS, job);
+ if (tableNames.length > 0)
+ return tableNames[0];
+ else
+ return null;
}
/**
@@ -329,7 +334,12 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #setRanges(JobConf, Collection)
*/
protected static List<Range> getRanges(JobConf job) throws IOException {
- return InputConfigurator.getRanges(CLASS, job);
+ Map<String, List<Range>> tableRanges = InputConfigurator.getRanges(CLASS,job);
+ List<Range> ranges = tableRanges.get(getInputTableName(job));
+ if(ranges != null)
+ return ranges;
+ else
+ return new LinkedList<Range>();
}
/**
@@ -356,7 +366,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #fetchColumns(JobConf, Collection)
*/
protected static Set<Pair<Text,Text>> getFetchedColumns(JobConf job) {
- return InputConfigurator.getFetchedColumns(CLASS, job);
+ return InputConfigurator.getFetchedColumns(CLASS, job, getInputTableName(job));
}
/**
@@ -382,7 +392,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @see #addIterator(JobConf, IteratorSetting)
*/
protected static List<IteratorSetting> getIterators(JobConf job) {
- return InputConfigurator.getIterators(CLASS, job);
+ return InputConfigurator.getDefaultIterators(CLASS,job);
}
/**
@@ -533,8 +543,8 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* if the table name set on the configuration doesn't exist
* @since 1.5.0
*/
- protected static TabletLocator getTabletLocator(JobConf job) throws TableNotFoundException {
- return InputConfigurator.getTabletLocator(CLASS, job);
+ protected static TabletLocator getTabletLocator(JobConf job, String tableName) throws TableNotFoundException {
+ return InputConfigurator.getTabletLocator(CLASS, job, tableName);
}
// InputFormat doesn't have the equivalent of OutputFormat's checkOutputSpecs(JobContext job)
@@ -760,9 +770,12 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
log.setLevel(getLogLevel(job));
validateOptions(job);
- String tableName = getInputTableName(job);
boolean autoAdjust = getAutoAdjustRanges(job);
- List<Range> ranges = autoAdjust ? Range.mergeOverlapping(getRanges(job)) : getRanges(job);
+ List<Range> tablesRanges = getRanges(job);
+ LinkedList<InputSplit> splits = new LinkedList<InputSplit>();
+
+ String tableName = getInputTableName(job);
+ List<Range> ranges = autoAdjust ? Range.mergeOverlapping(tablesRanges) : tablesRanges;
if (ranges.isEmpty()) {
ranges = new ArrayList<Range>(1);
@@ -772,6 +785,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
// get the metadata information for these ranges
Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>();
TabletLocator tl;
+ String tableId = null;
try {
if (isOfflineScan(job)) {
binnedRanges = binOfflineTable(job, tableName, ranges);
@@ -782,13 +796,13 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
}
} else {
Instance instance = getInstance(job);
- String tableId = null;
- tl = getTabletLocator(job);
+ tl = getTabletLocator(job, tableName);
// its possible that the cache could contain complete, but old information about a tables tablets... so clear it
tl.invalidateCache();
- while (!tl.binRanges(new Credentials(getPrincipal(job), getAuthenticationToken(job)), ranges, binnedRanges).isEmpty()) {
+ Credentials creds = new Credentials(getPrincipal(job), AuthenticationTokenSerializer.deserialize(getTokenClass(job), getToken(job)));
+
+ while (!tl.binRanges(creds, ranges, binnedRanges).isEmpty()) {
if (!(instance instanceof MockInstance)) {
- if (tableId == null)
tableId = Tables.getTableId(instance, tableName);
if (!Tables.exists(instance, tableId))
throw new TableDeletedException(tableId);
@@ -804,8 +818,6 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
} catch (Exception e) {
throw new IOException(e);
}
-
- ArrayList<InputSplit> splits = new ArrayList<InputSplit>(ranges.size());
HashMap<Range,ArrayList<String>> splitsToAdd = null;
if (!autoAdjust)
@@ -827,7 +839,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
for (Range r : extentRanges.getValue()) {
if (autoAdjust) {
// divide ranges into smaller ranges, based on the tablets
- splits.add(new RangeInputSplit(tableName, ke.clip(r), new String[] {location}));
+ splits.add(new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location}));
} else {
// don't divide ranges
ArrayList<String> locations = splitsToAdd.get(r);
@@ -842,7 +854,8 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
if (!autoAdjust)
for (Entry<Range,ArrayList<String>> entry : splitsToAdd.entrySet())
- splits.add(new RangeInputSplit(tableName, entry.getKey(), entry.getValue().toArray(new String[0])));
+ splits.add(new RangeInputSplit(tableName, tableId, entry.getKey(), entry.getValue().toArray(new String[0])));
+
return splits.toArray(new InputSplit[splits.size()]);
}
@@ -859,8 +872,8 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
super(split);
}
- protected RangeInputSplit(String table, Range range, String[] locations) {
- super(table, range, locations);
+ protected RangeInputSplit(String table, String tableId, Range range, String[] locations) {
+ super(table, tableId, range, locations);
}
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
index 13f9708..04ff718 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
@@ -16,6 +16,9 @@
*/
package org.apache.accumulo.core.client.mapreduce;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.accumulo.core.client.security.tokens.AuthenticationToken.AuthenticationTokenSerializer;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
@@ -26,6 +29,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -50,7 +54,7 @@ import org.apache.accumulo.core.client.impl.TabletLocator;
import org.apache.accumulo.core.client.mapreduce.lib.util.InputConfigurator;
import org.apache.accumulo.core.client.mock.MockInstance;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
-import org.apache.accumulo.core.client.security.tokens.AuthenticationToken.AuthenticationTokenSerializer;
+import org.apache.accumulo.core.conf.TableQueryConfig;
import org.apache.accumulo.core.data.ByteSequence;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.KeyExtent;
@@ -189,7 +193,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setConnectorInfo(Job, String, String)
*/
protected static AuthenticationToken getAuthenticationToken(JobContext context) {
- return InputConfigurator.getAuthenticationToken(CLASS, getConfiguration(context));
+ return InputConfigurator.getAuthenticationToken(CLASS,getConfiguration(context));
}
/**
@@ -231,7 +235,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setMockInstance(Job, String)
*/
protected static Instance getInstance(JobContext context) {
- return InputConfigurator.getInstance(CLASS, getConfiguration(context));
+ return InputConfigurator.getInstance(CLASS,getConfiguration(context));
}
/**
@@ -244,7 +248,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
*/
public static void setLogLevel(Job job, Level level) {
- InputConfigurator.setLogLevel(CLASS, job.getConfiguration(), level);
+ InputConfigurator.setLogLevel(CLASS,job.getConfiguration(),level);
}
/**
@@ -261,7 +265,8 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
}
/**
- * Sets the name of the input table, over which this job will scan.
+ * Sets the name of the input table, over which this job will scan. This method has been deprecated in favor of
+ * {@link InputFormatBase#setInputTableNames(org.apache.hadoop.mapreduce.Job, java.util.Collection)}
*
* @param job
* the Hadoop job instance to be configured
@@ -269,26 +274,28 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* the table to use when the tablename is null in the write call
* @since 1.5.0
*/
+ @Deprecated
public static void setInputTableName(Job job, String tableName) {
- InputConfigurator.setInputTableName(CLASS, job.getConfiguration(), tableName);
+ InputConfigurator.setInputTableName(CLASS,job.getConfiguration(),tableName);
}
/**
- * Gets the table name from the configuration.
+ * Sets the names of the input tables over which this job will scan.
*
- * @param context
- * the Hadoop context for the configured job
- * @return the table name
- * @since 1.5.0
- * @see #setInputTableName(Job, String)
+ * @param job
+ * the Hadoop job instance to be configured
+ * @param tableNames
+ * the table to use when the tablename is null in the write call
+ * @since 1.6.0
*/
- protected static String getInputTableName(JobContext context) {
- return InputConfigurator.getInputTableName(CLASS, getConfiguration(context));
+ @Deprecated
+ public static void setInputTableNames(Job job, Collection<String> tableNames) {
+ InputConfigurator.setInputTableNames(CLASS,job.getConfiguration(),tableNames);
}
/**
* Sets the {@link Authorizations} used to scan. Must be a subset of the user's authorization. Defaults to the empty set.
- *
+ *
* @param job
* the Hadoop job instance to be configured
* @param auths
@@ -296,12 +303,12 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
*/
public static void setScanAuthorizations(Job job, Authorizations auths) {
- InputConfigurator.setScanAuthorizations(CLASS, job.getConfiguration(), auths);
+ InputConfigurator.setScanAuthorizations(CLASS,job.getConfiguration(),auths);
}
/**
* Gets the authorizations to set for the scans from the configuration.
- *
+ *
* @param context
* the Hadoop context for the configured job
* @return the Accumulo scan authorizations
@@ -309,19 +316,35 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setScanAuthorizations(Job, Authorizations)
*/
protected static Authorizations getScanAuthorizations(JobContext context) {
- return InputConfigurator.getScanAuthorizations(CLASS, getConfiguration(context));
+ return InputConfigurator.getScanAuthorizations(CLASS,getConfiguration(context));
}
/**
- * Sets the input ranges to scan for this job. If not set, the entire table will be scanned.
- *
+ * Sets the input ranges to scan for all tables associated with this job. This will be added to any per-table ranges that have been set using
+ * {@link #setRanges(org.apache.hadoop.mapreduce.Job, java.util.Map)}
+ *
* @param job
* the Hadoop job instance to be configured
* @param ranges
* the ranges that will be mapped over
* @since 1.5.0
*/
+ @Deprecated
public static void setRanges(Job job, Collection<Range> ranges) {
+ InputConfigurator.setRanges(CLASS,job.getConfiguration(),ranges);
+ }
+
+ /**
+ * Sets the input ranges to scan per-table for this job.
+ *
+ * @param job
+ * the Hadoop job instance to be configured
+ * @param ranges
+ * the per-table ranges that will be mapped over
+ * @since 1.6.0
+ */
+ @Deprecated
+ public static void setRanges(Job job, Map<String,Collection<Range>> ranges) {
InputConfigurator.setRanges(CLASS, job.getConfiguration(), ranges);
}
@@ -335,13 +358,16 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* if the ranges have been encoded improperly
* @since 1.5.0
* @see #setRanges(Job, Collection)
+ * @see #setRanges(org.apache.hadoop.mapreduce.Job, java.util.Map)
*/
- protected static List<Range> getRanges(JobContext context) throws IOException {
+ @Deprecated
+ protected static Map<String,List<Range>> getRanges(JobContext context) throws IOException {
return InputConfigurator.getRanges(CLASS, getConfiguration(context));
}
/**
- * Restricts the columns that will be mapped over for this job.
+ * Restricts the columns that will be mapped over for this job for all tables. These columns will be added to any per-table columns set with
+ * {@link #fetchColumns(org.apache.hadoop.mapreduce.Job, java.util.Map)}.
*
* @param job
* the Hadoop job instance to be configured
@@ -350,25 +376,44 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* selected. An empty set is the default and is equivalent to scanning the all columns.
* @since 1.5.0
*/
+ @Deprecated
public static void fetchColumns(Job job, Collection<Pair<Text,Text>> columnFamilyColumnQualifierPairs) {
InputConfigurator.fetchColumns(CLASS, job.getConfiguration(), columnFamilyColumnQualifierPairs);
}
/**
- * Gets the columns to be mapped over from this job.
+ * Restricts the columns that will be mapped over for this job per table..
+ *
+ * @param job
+ * the Hadoop job instance to be configured
+ * @param columnFamilyColumnQualifierPairs
+ * A map keyed by table name where the value is a pair of {@link Text} objects corresponding to column family and column qualifier. If the column
+ * qualifier is null, the entire column family is selected. An empty set is the default and is equivalent to scanning the all columns.
+ * @since 1.6.0
+ */
+ @Deprecated
+ public static void fetchColumns(Job job, Map<String,Collection<Pair<Text,Text>>> columnFamilyColumnQualifierPairs) {
+ InputConfigurator.fetchColumns(CLASS,job.getConfiguration(),columnFamilyColumnQualifierPairs);
+ }
+
+ /**
+ * Gets the columns to be mapped over from this job. Any default columns as well as per-table columns will be returned.
*
* @param context
* the Hadoop context for the configured job
+ * @param table
+ * the table for which to return the columns
* @return a set of columns
- * @since 1.5.0
+ * @since 1.6.0
* @see #fetchColumns(Job, Collection)
*/
- protected static Set<Pair<Text,Text>> getFetchedColumns(JobContext context) {
- return InputConfigurator.getFetchedColumns(CLASS, getConfiguration(context));
+ @Deprecated
+ protected static Set<Pair<Text,Text>> getFetchedColumns(JobContext context, String table) {
+ return InputConfigurator.getFetchedColumns(CLASS,getConfiguration(context),table);
}
/**
- * Encode an iterator on the input for this job.
+ * Encode an iterator on the default all tables for this job.
*
* @param job
* the Hadoop job instance to be configured
@@ -376,12 +421,47 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* the configuration of the iterator
* @since 1.5.0
*/
+ @Deprecated
public static void addIterator(Job job, IteratorSetting cfg) {
- InputConfigurator.addIterator(CLASS, job.getConfiguration(), cfg);
+ InputConfigurator.addIterator(CLASS,job.getConfiguration(),cfg);
}
/**
- * Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration.
+ * Encode an iterator on the input for this job for the specified table.
+ *
+ * @param job
+ * the Hadoop job instance to be configured
+ * @param table
+ * the table for which to add the iterator
+ * @param cfg
+ * the configuration of the iterator
+ * @since 1.6.0
+ */
+ @Deprecated
+ public static void addIterator(Job job, String table, IteratorSetting cfg) {
+ InputConfigurator.addIterator(CLASS, job.getConfiguration(), table, cfg);
+ }
+
+ /**
+ * Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration for the specific table. Any default iterators will be
+ * included in the return.
+ *
+ * @param context
+ * the Hadoop context for the configured job
+ * @param table
+ * the table for which to return the iterators
+ * @return a list of iterators for the given table
+ * @since 1.6.0
+ * @see #addIterator(Job, String, IteratorSetting)
+ */
+ @Deprecated
+ protected static List<IteratorSetting> getIterators(JobContext context, String table) {
+ return InputConfigurator.getIterators(CLASS, getConfiguration(context), table);
+ }
+
+ /**
+ * Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration. This will only return iterators that have not been set
+ * for a specific table.
*
* @param context
* the Hadoop context for the configured job
@@ -389,8 +469,9 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
* @see #addIterator(Job, IteratorSetting)
*/
+ @Deprecated
protected static List<IteratorSetting> getIterators(JobContext context) {
- return InputConfigurator.getIterators(CLASS, getConfiguration(context));
+ return InputConfigurator.getDefaultIterators(CLASS,getConfiguration(context));
}
/**
@@ -407,6 +488,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setRanges(Job, Collection)
* @since 1.5.0
*/
+ @Deprecated
public static void setAutoAdjustRanges(Job job, boolean enableFeature) {
InputConfigurator.setAutoAdjustRanges(CLASS, job.getConfiguration(), enableFeature);
}
@@ -420,10 +502,25 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
* @see #setAutoAdjustRanges(Job, boolean)
*/
+ @Deprecated
protected static boolean getAutoAdjustRanges(JobContext context) {
return InputConfigurator.getAutoAdjustRanges(CLASS, getConfiguration(context));
}
+
+ protected static void setTableQueryConfigurations(JobContext job, TableQueryConfig... configs) {
+ checkNotNull(configs);
+ InputConfigurator.setTableQueryConfiguration(CLASS, getConfiguration(job), configs);
+ }
+
+ public static List<TableQueryConfig> getTableQueryConfigurations(JobContext job) {
+ return InputConfigurator.getTableQueryConfigurations(CLASS, getConfiguration(job));
+ }
+
+ protected static TableQueryConfig getTableQueryConfiguration(JobContext job, String tableName) {
+ return InputConfigurator.getTableQueryConfiguration(CLASS, getConfiguration(job), tableName);
+ }
+
/**
* Controls the use of the {@link IsolatedScanner} in this job.
*
@@ -536,13 +633,15 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
*
* @param context
* the Hadoop context for the configured job
+ * @param table
+ * the table for which to initialize the locator
* @return an Accumulo tablet locator
* @throws TableNotFoundException
* if the table name set on the configuration doesn't exist
* @since 1.5.0
*/
- protected static TabletLocator getTabletLocator(JobContext context) throws TableNotFoundException {
- return InputConfigurator.getTabletLocator(CLASS, getConfiguration(context));
+ protected static TabletLocator getTabletLocator(JobContext context, String table) throws TableNotFoundException {
+ return InputConfigurator.getTabletLocator(CLASS, getConfiguration(context), table);
}
// InputFormat doesn't have the equivalent of OutputFormat's checkOutputSpecs(JobContext job)
@@ -577,42 +676,60 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
protected RangeInputSplit split;
/**
- * Apply the configured iterators from the configuration to the scanner.
+ * Apply the configured iterators from the configuration to the scanner. This applies both the
+ * default iterators and the per-table iterators.
*
* @param context
* the Hadoop context for the configured job
* @param scanner
* the scanner to configure
+ * @param tableName
+ * the table name for which to set up the iterators
*/
- protected void setupIterators(TaskAttemptContext context, Scanner scanner) {
- List<IteratorSetting> iterators = getIterators(context);
- for (IteratorSetting iterator : iterators) {
+ protected void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName) {
+ List<IteratorSetting> iterators = getIterators(context, tableName); // default iterators will be included
+ for (IteratorSetting iterator : iterators)
scanner.addScanIterator(iterator);
- }
}
+
/**
* Initialize a scanner over the given input split using this task attempt configuration.
*/
@Override
public void initialize(InputSplit inSplit, TaskAttemptContext attempt) throws IOException {
+
Scanner scanner;
split = (RangeInputSplit) inSplit;
- log.debug("Initializing input split: " + split.range);
+ log.debug("Initializing input split: " + split.getRange());
Instance instance = getInstance(attempt);
String principal = getPrincipal(attempt);
+
+ TableQueryConfig tableConfig = getTableQueryConfiguration(attempt, split.getTableName());
+
+ // in case the table name changed, we can still use the previous name for terms of configuration,
+ // but for the scanner, we'll need to reference the new table name.
+ String actualNameForId = null;
+ try{
+ actualNameForId = Tables.getTableName(instance, split.getTableId());
+ if(!actualNameForId.equals(split.getTableName()))
+ log.debug("Table name changed from " + split.getTableName() + " to " + actualNameForId);
+ }catch(TableNotFoundException e){
+ throw new IOException("The specified table was not found for id=" + split.getTableId());
+ }
+
AuthenticationToken token = getAuthenticationToken(attempt);
Authorizations authorizations = getScanAuthorizations(attempt);
-
try {
log.debug("Creating connector with user: " + principal);
+
Connector conn = instance.getConnector(principal, token);
- log.debug("Creating scanner for table: " + getInputTableName(attempt));
+ log.debug("Creating scanner for table: " + actualNameForId);
log.debug("Authorizations are: " + authorizations);
if (isOfflineScan(attempt)) {
- scanner = new OfflineScanner(instance, new Credentials(principal, token), Tables.getTableId(instance, getInputTableName(attempt)), authorizations);
+ scanner = new OfflineScanner(instance, new Credentials(principal, token), split.getTableId(), authorizations);
} else {
- scanner = conn.createScanner(getInputTableName(attempt), authorizations);
+ scanner = conn.createScanner(actualNameForId, authorizations);
}
if (isIsolated(attempt)) {
log.info("Creating isolated scanner");
@@ -622,13 +739,13 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
log.info("Using local iterators");
scanner = new ClientSideIteratorScanner(scanner);
}
- setupIterators(attempt, scanner);
+ setupIterators(attempt, scanner, split.getTableName());
} catch (Exception e) {
throw new IOException(e);
}
// setup a scanner within the bounds of this split
- for (Pair<Text,Text> c : getFetchedColumns(attempt)) {
+ for (Pair<Text,Text> c : tableConfig.getColumns()) {
if (c.getSecond() != null) {
log.debug("Fetching column " + c.getFirst() + ":" + c.getSecond());
scanner.fetchColumn(c.getFirst(), c.getSecond());
@@ -638,8 +755,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
}
}
- scanner.setRange(split.range);
-
+ scanner.setRange(split.getRange());
numKeysRead = 0;
// do this last after setting all scanner options
@@ -705,9 +821,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
scanner.setRange(metadataRange);
RowIterator rowIter = new RowIterator(scanner);
-
KeyExtent lastExtent = null;
-
while (rowIter.hasNext()) {
Iterator<Entry<Key,Value>> row = rowIter.next();
String last = "";
@@ -771,96 +885,104 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
}
/**
- * Read the metadata table to get tablets and match up ranges to them.
+ * Gets the splits of the tables that have been set on the job.
+ *
+ * @param conf
+ * the configuration of the job
+ * @return the splits from the tables based on the ranges.
+ * @throws IOException
+ * if a table set on the job doesn't exist or an error occurs
+ * initializing the tablet locator
*/
- @Override
- public List<InputSplit> getSplits(JobContext context) throws IOException {
- log.setLevel(getLogLevel(context));
- validateOptions(context);
-
- String tableName = getInputTableName(context);
- boolean autoAdjust = getAutoAdjustRanges(context);
- List<Range> ranges = autoAdjust ? Range.mergeOverlapping(getRanges(context)) : getRanges(context);
+ public List<InputSplit> getSplits(JobContext conf) throws IOException {
+ log.setLevel(getLogLevel(conf));
+ validateOptions(conf);
+
+ LinkedList<InputSplit> splits = new LinkedList<InputSplit>();
+ List<TableQueryConfig> tableConfigs = getTableQueryConfigurations(conf);
+ for (TableQueryConfig tableConfig : tableConfigs) {
+
+ boolean autoAdjust = getAutoAdjustRanges(conf); // TODO: Put this in the table config object
+ String tableId = null;
+ List<Range> ranges = autoAdjust ? Range.mergeOverlapping(tableConfig.getRanges()) : tableConfig.getRanges();
+ if (ranges.isEmpty()) {
+ ranges = new ArrayList<Range>(1);
+ ranges.add(new Range());
+ }
- if (ranges.isEmpty()) {
- ranges = new ArrayList<Range>(1);
- ranges.add(new Range());
- }
+ // get the metadata information for these ranges
+ Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>();
+ TabletLocator tl;
+ try {
+ if (isOfflineScan(conf)) {
+ binnedRanges = binOfflineTable(conf, tableConfig.getTableName(), ranges);
+ while (binnedRanges == null) {
+ // Some tablets were still online, try again
+ UtilWaitThread.sleep(100 + (int) (Math.random() * 100)); // sleep randomly between 100 and 200 ms
+ binnedRanges = binOfflineTable(conf, tableConfig.getTableName(), ranges);
- // get the metadata information for these ranges
- Map<String,Map<KeyExtent,List<Range>>> binnedRanges = new HashMap<String,Map<KeyExtent,List<Range>>>();
- TabletLocator tl;
- try {
- if (isOfflineScan(context)) {
- binnedRanges = binOfflineTable(context, tableName, ranges);
- while (binnedRanges == null) {
- // Some tablets were still online, try again
- UtilWaitThread.sleep(100 + (int) (Math.random() * 100)); // sleep randomly between 100 and 200 ms
- binnedRanges = binOfflineTable(context, tableName, ranges);
- }
- } else {
- Instance instance = getInstance(context);
- String tableId = null;
- tl = getTabletLocator(context);
- // its possible that the cache could contain complete, but old information about a tables tablets... so clear it
- tl.invalidateCache();
- while (!tl.binRanges(new Credentials(getPrincipal(context), getAuthenticationToken(context)), ranges, binnedRanges).isEmpty()) {
- if (!(instance instanceof MockInstance)) {
- if (tableId == null)
- tableId = Tables.getTableId(instance, tableName);
- if (!Tables.exists(instance, tableId))
- throw new TableDeletedException(tableId);
- if (Tables.getTableState(instance, tableId) == TableState.OFFLINE)
- throw new TableOfflineException(instance, tableId);
}
- binnedRanges.clear();
- log.warn("Unable to locate bins for specified ranges. Retrying.");
- UtilWaitThread.sleep(100 + (int) (Math.random() * 100)); // sleep randomly between 100 and 200 ms
+ } else {
+ Instance instance = getInstance(conf);
+ tl = getTabletLocator(conf, tableConfig.getTableName());
+ // its possible that the cache could contain complete, but old information about a tables tablets... so clear it
tl.invalidateCache();
+ Credentials creds = new Credentials(getPrincipal(conf), getAuthenticationToken(conf));
+
+ while (!tl.binRanges(creds, ranges, binnedRanges).isEmpty()) {
+ if (!(instance instanceof MockInstance)) {
+ if (!Tables.exists(instance, tableId))
+ throw new TableDeletedException(tableId);
+ if (Tables.getTableState(instance, tableId) == TableState.OFFLINE)
+ throw new TableOfflineException(instance, tableId);
+ tableId = Tables.getTableId(instance, tableConfig.getTableName());
+ }
+ binnedRanges.clear();
+ log.warn("Unable to locate bins for specified ranges. Retrying.");
+ UtilWaitThread.sleep(100 + (int) (Math.random() * 100)); // sleep randomly between 100 and 200 ms
+ tl.invalidateCache();
+ }
}
+ } catch (Exception e) {
+ throw new IOException(e);
}
- } catch (Exception e) {
- throw new IOException(e);
- }
-
- ArrayList<InputSplit> splits = new ArrayList<InputSplit>(ranges.size());
- HashMap<Range,ArrayList<String>> splitsToAdd = null;
- if (!autoAdjust)
- splitsToAdd = new HashMap<Range,ArrayList<String>>();
+ HashMap<Range,ArrayList<String>> splitsToAdd = null;
- HashMap<String,String> hostNameCache = new HashMap<String,String>();
+ if (!autoAdjust)
+ splitsToAdd = new HashMap<Range,ArrayList<String>>();
- for (Entry<String,Map<KeyExtent,List<Range>>> tserverBin : binnedRanges.entrySet()) {
- String ip = tserverBin.getKey().split(":", 2)[0];
- String location = hostNameCache.get(ip);
- if (location == null) {
- InetAddress inetAddress = InetAddress.getByName(ip);
- location = inetAddress.getHostName();
- hostNameCache.put(ip, location);
- }
-
- for (Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
- Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- splits.add(new RangeInputSplit(tableName, ke.clip(r), new String[] {location}));
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ HashMap<String,String> hostNameCache = new HashMap<String,String>();
+ for (Entry<String,Map<KeyExtent,List<Range>>> tserverBin : binnedRanges.entrySet()) {
+ String ip = tserverBin.getKey().split(":", 2)[0];
+ String location = hostNameCache.get(ip);
+ if (location == null) {
+ InetAddress inetAddress = InetAddress.getByName(ip);
+ location = inetAddress.getHostName();
+ hostNameCache.put(ip, location);
+ }
+ for (Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
+ Range ke = extentRanges.getKey().toDataRange();
+ for (Range r : extentRanges.getValue()) {
+ if (autoAdjust) {
+ // divide ranges into smaller ranges, based on the tablets
+ splits.add(new RangeInputSplit(tableConfig.getTableName(), tableId, ke.clip(r), new String[] {location}));
+ } else {
+ // don't divide ranges
+ ArrayList<String> locations = splitsToAdd.get(r);
+ if (locations == null)
+ locations = new ArrayList<String>(1);
+ locations.add(location);
+ splitsToAdd.put(r, locations);
+ }
}
}
}
- }
- if (!autoAdjust)
- for (Entry<Range,ArrayList<String>> entry : splitsToAdd.entrySet())
- splits.add(new RangeInputSplit(tableName, entry.getKey(), entry.getValue().toArray(new String[0])));
+ if (!autoAdjust)
+ for (Entry<Range,ArrayList<String>> entry : splitsToAdd.entrySet())
+ splits.add(new RangeInputSplit(tableConfig.getTableName(), tableId, entry.getKey(), entry.getValue().toArray(new String[0])));
+ }
return splits;
}
@@ -870,20 +992,27 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
public static class RangeInputSplit extends InputSplit implements Writable {
private Range range;
private String[] locations;
+ private String tableId;
+ private String tableName;
public RangeInputSplit() {
range = new Range();
locations = new String[0];
+ tableId = "";
+ tableName = "";
}
public RangeInputSplit(RangeInputSplit split) throws IOException {
this.setRange(split.getRange());
this.setLocations(split.getLocations());
+ this.setTableName(split.getTableName());
}
- protected RangeInputSplit(String table, Range range, String[] locations) {
+ protected RangeInputSplit(String table, String tableId, Range range, String[] locations) {
this.range = range;
this.locations = locations;
+ this.tableName = table;
+ this.tableId = tableId;
}
public Range getRange() {
@@ -894,6 +1023,22 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
this.range = range;
}
+ public String getTableName() {
+ return tableName;
+ }
+
+ public void setTableName(String tableName) {
+ this.tableName = tableName;
+ }
+
+ public void setTableId(String tableId) {
+ this.tableId = tableId;
+ }
+
+ public String getTableId() {
+ return tableId;
+ }
+
private static byte[] extractBytes(ByteSequence seq, int numBytes) {
byte[] bytes = new byte[numBytes + 1];
bytes[0] = 0;
@@ -968,6 +1113,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
@Override
public void readFields(DataInput in) throws IOException {
range.readFields(in);
+ tableName = in.readUTF();
int numLocs = in.readInt();
locations = new String[numLocs];
for (int i = 0; i < numLocs; ++i)
@@ -977,6 +1123,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
@Override
public void write(DataOutput out) throws IOException {
range.write(out);
+ out.writeUTF(tableName);
out.writeInt(locations.length);
for (int i = 0; i < locations.length; ++i)
out.writeUTF(locations[i]);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
index a1c3f70..29795af 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
@@ -16,6 +16,8 @@
*/
package org.apache.accumulo.core.client.mapreduce.lib.util;
+import static org.apache.accumulo.core.util.ArgumentChecker.notNull;
+
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
@@ -23,10 +25,8 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashSet;
import java.util.List;
-import java.util.Set;
-import java.util.StringTokenizer;
+import java.util.TreeMap;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
@@ -42,11 +42,11 @@ import org.apache.accumulo.core.client.impl.Tables;
import org.apache.accumulo.core.client.impl.TabletLocator;
import org.apache.accumulo.core.client.mock.MockTabletLocator;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
+import org.apache.accumulo.core.conf.TableQueryConfig;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.TablePermission;
-import org.apache.accumulo.core.util.ArgumentChecker;
import org.apache.accumulo.core.util.Pair;
import org.apache.accumulo.core.util.TextUtil;
import org.apache.commons.codec.binary.Base64;
@@ -65,7 +65,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @since 1.5.0
*/
public static enum ScanOpts {
- TABLE_NAME, AUTHORIZATIONS, RANGES, COLUMNS, ITERATORS
+ TABLE, AUTHORIZATIONS, RANGES, COLUMNS, ITERATORS
}
/**
@@ -87,25 +87,27 @@ public class InputConfigurator extends ConfiguratorBase {
* @param tableName
* the table to use when the tablename is null in the write call
* @since 1.5.0
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static void setInputTableName(Class<?> implementingClass, Configuration conf, String tableName) {
- ArgumentChecker.notNull(tableName);
- conf.set(enumToConfKey(implementingClass, ScanOpts.TABLE_NAME), tableName);
+ notNull(tableName);
+ conf.set(enumToConfKey(implementingClass, ScanOpts.TABLE), tableName);
}
/**
- * Gets the table name from the configuration.
- *
+ * Sets the name of the input table, over which this job will scan.
+ *
* @param implementingClass
* the class whose name will be used as a prefix for the property configuration key
* @param conf
* the Hadoop configuration object to configure
- * @return the table name
* @since 1.5.0
- * @see #setInputTableName(Class, Configuration, String)
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static String getInputTableName(Class<?> implementingClass, Configuration conf) {
- return conf.get(enumToConfKey(implementingClass, ScanOpts.TABLE_NAME));
+ return conf.get(enumToConfKey(implementingClass, ScanOpts.TABLE));
}
/**
@@ -141,7 +143,7 @@ public class InputConfigurator extends ConfiguratorBase {
}
/**
- * Sets the input ranges to scan for this job. If not set, the entire table will be scanned.
+ * Sets the input ranges to scan on all input tables for this job. If not set, the entire table will be scanned.
*
* @param implementingClass
* the class whose name will be used as a prefix for the property configuration key
@@ -149,21 +151,26 @@ public class InputConfigurator extends ConfiguratorBase {
* the Hadoop configuration object to configure
* @param ranges
* the ranges that will be mapped over
+ * @throws IllegalArgumentException
+ * if the ranges cannot be encoded into base 64
* @since 1.5.0
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static void setRanges(Class<?> implementingClass, Configuration conf, Collection<Range> ranges) {
- ArgumentChecker.notNull(ranges);
+ notNull(ranges);
+
ArrayList<String> rangeStrings = new ArrayList<String>(ranges.size());
try {
for (Range r : ranges) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
r.write(new DataOutputStream(baos));
- rangeStrings.add(new String(Base64.encodeBase64(baos.toByteArray()), Constants.UTF8));
+ rangeStrings.add(new String(Base64.encodeBase64(baos.toByteArray())));
}
+ conf.setStrings(enumToConfKey(implementingClass, ScanOpts.RANGES), rangeStrings.toArray(new String[0]));
} catch (IOException ex) {
throw new IllegalArgumentException("Unable to encode ranges to Base64", ex);
}
- conf.setStrings(enumToConfKey(implementingClass, ScanOpts.RANGES), rangeStrings.toArray(new String[0]));
}
/**
@@ -176,13 +183,17 @@ public class InputConfigurator extends ConfiguratorBase {
* @return the ranges
* @throws IOException
* if the ranges have been encoded improperly
- * @since 1.5.0
+ * @since 1.6.0
+ * @deprecated since 1.6.0
* @see #setRanges(Class, Configuration, Collection)
*/
+ @Deprecated
public static List<Range> getRanges(Class<?> implementingClass, Configuration conf) throws IOException {
- ArrayList<Range> ranges = new ArrayList<Range>();
- for (String rangeString : conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.RANGES))) {
- ByteArrayInputStream bais = new ByteArrayInputStream(Base64.decodeBase64(rangeString.getBytes(Constants.UTF8)));
+
+ Collection<String> encodedRanges = conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.RANGES));
+ List<Range> ranges = new ArrayList<Range>();
+ for (String rangeString : encodedRanges) {
+ ByteArrayInputStream bais = new ByteArrayInputStream(Base64.decodeBase64(rangeString.getBytes()));
Range range = new Range();
range.readFields(new DataInputStream(bais));
ranges.add(range);
@@ -191,7 +202,7 @@ public class InputConfigurator extends ConfiguratorBase {
}
/**
- * Restricts the columns that will be mapped over for this job.
+ * Restricts the columns that will be mapped over for this job. This applies the columns to all tables that have been set on the job.
*
* @param implementingClass
* the class whose name will be used as a prefix for the property configuration key
@@ -200,12 +211,17 @@ public class InputConfigurator extends ConfiguratorBase {
* @param columnFamilyColumnQualifierPairs
* a pair of {@link Text} objects corresponding to column family and column qualifier. If the column qualifier is null, the entire column family is
* selected. An empty set is the default and is equivalent to scanning the all columns.
+ * @throws IllegalArgumentException
+ * if the column family is null
* @since 1.5.0
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static void fetchColumns(Class<?> implementingClass, Configuration conf, Collection<Pair<Text,Text>> columnFamilyColumnQualifierPairs) {
- ArgumentChecker.notNull(columnFamilyColumnQualifierPairs);
- ArrayList<String> columnStrings = new ArrayList<String>(columnFamilyColumnQualifierPairs.size());
+ notNull(columnFamilyColumnQualifierPairs);
+ ArrayList<String> columnStrings = new ArrayList<String>();
for (Pair<Text,Text> column : columnFamilyColumnQualifierPairs) {
+
if (column.getFirst() == null)
throw new IllegalArgumentException("Column family can not be null");
@@ -218,29 +234,7 @@ public class InputConfigurator extends ConfiguratorBase {
}
/**
- * Gets the columns to be mapped over from this job.
- *
- * @param implementingClass
- * the class whose name will be used as a prefix for the property configuration key
- * @param conf
- * the Hadoop configuration object to configure
- * @return a set of columns
- * @since 1.5.0
- * @see #fetchColumns(Class, Configuration, Collection)
- */
- public static Set<Pair<Text,Text>> getFetchedColumns(Class<?> implementingClass, Configuration conf) {
- Set<Pair<Text,Text>> columns = new HashSet<Pair<Text,Text>>();
- for (String col : conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.COLUMNS))) {
- int idx = col.indexOf(":");
- Text cf = new Text(idx < 0 ? Base64.decodeBase64(col.getBytes(Constants.UTF8)) : Base64.decodeBase64(col.substring(0, idx).getBytes(Constants.UTF8)));
- Text cq = idx < 0 ? null : new Text(Base64.decodeBase64(col.substring(idx + 1).getBytes()));
- columns.add(new Pair<Text,Text>(cf, cq));
- }
- return columns;
- }
-
- /**
- * Encode an iterator on the input for this job.
+ * Encode an iterator on the input for all tables associated with this job.
*
* @param implementingClass
* the class whose name will be used as a prefix for the property configuration key
@@ -248,8 +242,12 @@ public class InputConfigurator extends ConfiguratorBase {
* the Hadoop configuration object to configure
* @param cfg
* the configuration of the iterator
+ * @throws IllegalArgumentException
+ * if the iterator can't be serialized into the configuration
* @since 1.5.0
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static void addIterator(Class<?> implementingClass, Configuration conf, IteratorSetting cfg) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
String newIter;
@@ -261,7 +259,8 @@ public class InputConfigurator extends ConfiguratorBase {
throw new IllegalArgumentException("unable to serialize IteratorSetting");
}
- String iterators = conf.get(enumToConfKey(implementingClass, ScanOpts.ITERATORS));
+ String confKey = enumToConfKey(implementingClass, ScanOpts.ITERATORS);
+ String iterators = conf.get(confKey);
// No iterators specified yet, create a new string
if (iterators == null || iterators.isEmpty()) {
iterators = newIter;
@@ -270,41 +269,7 @@ public class InputConfigurator extends ConfiguratorBase {
iterators = iterators.concat(StringUtils.COMMA_STR + newIter);
}
// Store the iterators w/ the job
- conf.set(enumToConfKey(implementingClass, ScanOpts.ITERATORS), iterators);
- }
-
- /**
- * Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration.
- *
- * @param implementingClass
- * the class whose name will be used as a prefix for the property configuration key
- * @param conf
- * the Hadoop configuration object to configure
- * @return a list of iterators
- * @since 1.5.0
- * @see #addIterator(Class, Configuration, IteratorSetting)
- */
- public static List<IteratorSetting> getIterators(Class<?> implementingClass, Configuration conf) {
- String iterators = conf.get(enumToConfKey(implementingClass, ScanOpts.ITERATORS));
-
- // If no iterators are present, return an empty list
- if (iterators == null || iterators.isEmpty())
- return new ArrayList<IteratorSetting>();
-
- // Compose the set of iterators encoded in the job configuration
- StringTokenizer tokens = new StringTokenizer(iterators, StringUtils.COMMA_STR);
- List<IteratorSetting> list = new ArrayList<IteratorSetting>();
- try {
- while (tokens.hasMoreTokens()) {
- String itstring = tokens.nextToken();
- ByteArrayInputStream bais = new ByteArrayInputStream(Base64.decodeBase64(itstring.getBytes()));
- list.add(new IteratorSetting(new DataInputStream(bais)));
- bais.close();
- }
- } catch (IOException e) {
- throw new IllegalArgumentException("couldn't decode iterator settings");
- }
- return list;
+ conf.set(confKey, iterators);
}
/**
@@ -322,7 +287,9 @@ public class InputConfigurator extends ConfiguratorBase {
* the feature is enabled if true, disabled otherwise
* @see #setRanges(Class, Configuration, Collection)
* @since 1.5.0
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static void setAutoAdjustRanges(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
conf.setBoolean(enumToConfKey(implementingClass, Features.AUTO_ADJUST_RANGES), enableFeature);
}
@@ -337,7 +304,9 @@ public class InputConfigurator extends ConfiguratorBase {
* @return false if the feature is disabled, true otherwise
* @since 1.5.0
* @see #setAutoAdjustRanges(Class, Configuration, boolean)
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static Boolean getAutoAdjustRanges(Class<?> implementingClass, Configuration conf) {
return conf.getBoolean(enumToConfKey(implementingClass, Features.AUTO_ADJUST_RANGES), true);
}
@@ -355,7 +324,9 @@ public class InputConfigurator extends ConfiguratorBase {
* @param enableFeature
* the feature is enabled if true, disabled otherwise
* @since 1.5.0
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static void setScanIsolation(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
conf.setBoolean(enumToConfKey(implementingClass, Features.SCAN_ISOLATION), enableFeature);
}
@@ -370,7 +341,9 @@ public class InputConfigurator extends ConfiguratorBase {
* @return true if the feature is enabled, false otherwise
* @since 1.5.0
* @see #setScanIsolation(Class, Configuration, boolean)
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static Boolean isIsolated(Class<?> implementingClass, Configuration conf) {
return conf.getBoolean(enumToConfKey(implementingClass, Features.SCAN_ISOLATION), false);
}
@@ -389,7 +362,9 @@ public class InputConfigurator extends ConfiguratorBase {
* @param enableFeature
* the feature is enabled if true, disabled otherwise
* @since 1.5.0
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static void setLocalIterators(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
conf.setBoolean(enumToConfKey(implementingClass, Features.USE_LOCAL_ITERATORS), enableFeature);
}
@@ -404,7 +379,9 @@ public class InputConfigurator extends ConfiguratorBase {
* @return true if the feature is enabled, false otherwise
* @since 1.5.0
* @see #setLocalIterators(Class, Configuration, boolean)
+ * @deprecated since 1.6.0
*/
+ @Deprecated
public static Boolean usesLocalIterators(Class<?> implementingClass, Configuration conf) {
return conf.getBoolean(enumToConfKey(implementingClass, Features.USE_LOCAL_ITERATORS), false);
}
@@ -461,6 +438,59 @@ public class InputConfigurator extends ConfiguratorBase {
return conf.getBoolean(enumToConfKey(implementingClass, Features.SCAN_OFFLINE), false);
}
+ public static void setTableQueryConfiguration(Class<?> implementingClass, Configuration conf, TableQueryConfig... tconf) {
+ List<String> tableQueryConfigStrings = new ArrayList<String>();
+ for(TableQueryConfig queryConfig : tconf) {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try {
+ queryConfig.write(new DataOutputStream(baos));
+ } catch(IOException e) {
+ throw new IllegalStateException("Configuration for " + queryConfig.getTableName() + " could not be serialized.");
+ }
+ tableQueryConfigStrings.add(new String(Base64.encodeBase64(baos.toByteArray())));
+ }
+ String confKey = enumToConfKey(implementingClass, ScanOpts.TABLE);
+ conf.setStrings(confKey, tableQueryConfigStrings.toArray(new String[0]));
+ }
+
+ public static List<TableQueryConfig> getTableQueryConfigurations(Class<?> implementingClass, Configuration conf) {
+ List<TableQueryConfig> configs = new ArrayList<TableQueryConfig>();
+ Collection<String> configStrings = conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.TABLE));
+ if(configStrings == null) {
+ for(String str : configStrings) {
+ try{
+ byte[] bytes = Base64.decodeBase64(str.getBytes());
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ configs.add(new TableQueryConfig(new DataInputStream(bais)));
+ bais.close();
+ } catch(IOException e) {
+ throw new IllegalStateException("The table query configurations could not be deserialized from the given configuration");
+ }
+ }
+ }
+ return configs;
+ }
+
+ public static TableQueryConfig getTableQueryConfiguration(Class<?> implementingClass, Configuration conf, String tableName) {
+ Collection<String> configStrings = conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.TABLE));
+ if(configStrings == null) {
+ for(String str : configStrings) {
+ if(str.equals(tableName)) {
+ try{
+ byte[] bytes = Base64.decodeBase64(str.getBytes());
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ TableQueryConfig config = new TableQueryConfig(new DataInputStream(bais));
+ bais.close();
+ return config;
+ } catch(IOException e) {
+ throw new IllegalStateException("The table query configurations could not be deserialized from the given configuration");
+ }
+ }
+ }
+ }
+ return null;
+ }
+
/**
* Initializes an Accumulo {@link TabletLocator} based on the configuration.
*
@@ -468,17 +498,18 @@ public class InputConfigurator extends ConfiguratorBase {
* the class whose name will be used as a prefix for the property configuration key
* @param conf
* the Hadoop configuration object to configure
+ * @param tableName
+ * The table name for which to initialize the {@link TabletLocator}
* @return an Accumulo tablet locator
* @throws TableNotFoundException
* if the table name set on the configuration doesn't exist
- * @since 1.5.0
+ * @since 1.6.0
*/
- public static TabletLocator getTabletLocator(Class<?> implementingClass, Configuration conf) throws TableNotFoundException {
+ public static TabletLocator getTabletLocator(Class<?> implementingClass, Configuration conf, String tableName) throws TableNotFoundException {
String instanceType = conf.get(enumToConfKey(implementingClass, InstanceOpts.TYPE));
if ("MockInstance".equals(instanceType))
return new MockTabletLocator();
Instance instance = getInstance(implementingClass, conf);
- String tableName = getInputTableName(implementingClass, conf);
return TabletLocator.getLocator(instance, new Text(Tables.getTableId(instance, tableName)));
}
@@ -507,17 +538,26 @@ public class InputConfigurator extends ConfiguratorBase {
Connector c = getInstance(implementingClass, conf).getConnector(principal, token);
if (!c.securityOperations().authenticateUser(principal, token))
throw new IOException("Unable to authenticate user");
- if (!c.securityOperations().hasTablePermission(principal, getInputTableName(implementingClass, conf), TablePermission.READ))
- throw new IOException("Unable to access table");
-
- if (!conf.getBoolean(enumToConfKey(implementingClass, Features.USE_LOCAL_ITERATORS), false)) {
- // validate that any scan-time iterators can be loaded by the the tablet servers
- for (IteratorSetting iter : getIterators(implementingClass, conf)) {
- if (!c.tableOperations().testClassLoad(getInputTableName(implementingClass, conf), iter.getIteratorClass(), SortedKeyValueIterator.class.getName()))
- throw new AccumuloException("Servers are unable to load " + iter.getIteratorClass() + " as a " + SortedKeyValueIterator.class.getName());
+
+ for (TableQueryConfig tableConfig : getTableQueryConfigurations(implementingClass, conf)) {
+ if (!c.securityOperations().hasTablePermission(getPrincipal(implementingClass, conf), tableConfig.getTableName(), TablePermission.READ))
+ throw new IOException("Unable to access table");
+ }
+
+ for (TableQueryConfig tableConfig : getTableQueryConfigurations(implementingClass,conf)) {
+ if(!tableConfig.shouldUseLocalIterators()) {
+// if (!conf.getBoolean(enumToConfKey(implementingClass, Features.USE_LOCAL_ITERATORS), false)) {
+ // validate that any scan-time iterators can be loaded by the the tablet servers
+ for (IteratorSetting iter : tableConfig.getIterators()) { // TODO: These iterators need to be separated by table
+ if (!c.tableOperations().testClassLoad(tableConfig.getTableName(), iter.getIteratorClass(), SortedKeyValueIterator.class.getName()))
+ throw new AccumuloException("Servers are unable to load " + iter.getIteratorClass() + " as a " + SortedKeyValueIterator.class.getName());
+
+ }
}
}
+ // TODO: Check for the "default table case"
+
} catch (AccumuloException e) {
throw new IOException(e);
} catch (AccumuloSecurityException e) {
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java b/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java
new file mode 100644
index 0000000..00f4dc0
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/conf/TableQueryConfig.java
@@ -0,0 +1,199 @@
+package org.apache.accumulo.core.conf;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.util.Pair;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+
+public class TableQueryConfig implements Writable {
+
+ private String tableName;
+ private List<IteratorSetting> iterators;
+ private List<Range> ranges;
+ private Set<Pair<Text,Text>> columns;
+
+ private boolean autoAdjustRanges = true;
+ private boolean useLocalIterators = false;
+ private boolean useIsolatedScanners = false;
+
+ public TableQueryConfig(String tableName) {
+ checkNotNull(tableName);
+ this.tableName = tableName;
+ }
+
+ public TableQueryConfig(DataInput input) throws IOException{
+ readFields(input);
+ }
+
+ public TableQueryConfig setRanges(List<Range> ranges) {
+ this.ranges = ranges;
+ return this;
+ }
+
+ public TableQueryConfig setColumns(Set<Pair<Text,Text>> columns) {
+ this.columns = columns;
+ return this;
+ }
+
+ public TableQueryConfig setIterators(List<IteratorSetting> iterators) {
+ this.iterators = iterators;
+ return this;
+ }
+
+ public TableQueryConfig setAutoAdjustRanges(boolean autoAdjustRanges){
+ this.autoAdjustRanges=autoAdjustRanges;
+ return this;
+ }
+
+ public TableQueryConfig setUseLocalIterators(boolean useLocalIterators){
+ this.useLocalIterators=useLocalIterators;
+ return this;
+ }
+
+ public TableQueryConfig setUseIsolatedScanners(boolean useIsolatedScanners){
+ this.useIsolatedScanners=useIsolatedScanners;
+ return this;
+ }
+
+ public String getTableName(){
+ return tableName;
+ }
+
+ public List<IteratorSetting> getIterators(){
+ return iterators;
+ }
+
+ public List<Range> getRanges(){
+ return ranges;
+ }
+
+ public Set<Pair<Text,Text>> getColumns(){
+ return columns;
+ }
+
+ public boolean shouldAutoAdjustRanges(){
+ return autoAdjustRanges;
+ }
+
+ public boolean shouldUseLocalIterators(){
+ return useLocalIterators;
+ }
+
+ public boolean shouldUseIsolatedScanners(){
+ return useIsolatedScanners;
+ }
+
+ @Override
+ public void write(DataOutput dataOutput) throws IOException {
+ dataOutput.writeUTF(tableName);
+ if (iterators != null) {
+ dataOutput.writeInt(iterators.size());
+ for (IteratorSetting setting : iterators)
+ setting.write(dataOutput);
+ } else {
+ dataOutput.writeInt(0);
+ }
+ if (ranges != null) {
+ dataOutput.writeInt(ranges.size());
+ for (Range range : ranges)
+ range.write(dataOutput);
+ } else {
+ dataOutput.writeInt(0);
+ }
+ if (columns != null) {
+ dataOutput.writeInt(columns.size());
+ for (Pair<Text,Text> column : columns) {
+ if (column.getSecond() == null) {
+ dataOutput.writeInt(1);
+ column.getFirst().write(dataOutput);
+ } else {
+ dataOutput.writeInt(2);
+ column.getFirst().write(dataOutput);
+ column.getSecond().write(dataOutput);
+ }
+ }
+ } else {
+ dataOutput.writeInt(0);
+ }
+ dataOutput.writeBoolean(autoAdjustRanges);
+ dataOutput.writeBoolean(useLocalIterators);
+ dataOutput.writeBoolean(useIsolatedScanners);
+ }
+
+ @Override
+ public void readFields(DataInput dataInput) throws IOException {
+ this.tableName = dataInput.readUTF();
+ // load iterators
+ long iterSize = dataInput.readInt();
+ if (iterSize > 0)
+ iterators = new ArrayList<IteratorSetting>();
+ for (int i = 0; i < iterSize; i++)
+ iterators.add(new IteratorSetting(dataInput));
+ // load ranges
+ long rangeSize = dataInput.readInt();
+ if (rangeSize > 0)
+ ranges = new ArrayList<Range>();
+ for (int i = 0; i < rangeSize; i++) {
+ Range range = new Range();
+ range.readFields(dataInput);
+ ranges.add(range);
+ }
+ // load columns
+ long columnSize = dataInput.readInt();
+ if (columnSize > 0)
+ columns = new HashSet<Pair<Text,Text>>();
+ for (int i = 0; i < columnSize; i++) {
+ long numPairs = dataInput.readInt();
+ Text colFam = new Text();
+ colFam.readFields(dataInput);
+ if (numPairs == 1) {
+ columns.add(new Pair<Text,Text>(colFam, null));
+ } else if (numPairs == 2) {
+ Text colQual = new Text();
+ colQual.readFields(dataInput);
+ columns.add(new Pair<Text,Text>(colFam, colQual));
+ }
+ }
+ autoAdjustRanges = dataInput.readBoolean();
+ useLocalIterators = dataInput.readBoolean();
+ useIsolatedScanners = dataInput.readBoolean();
+ }
+
+ @Override
+ public boolean equals(Object o){
+ if(this==o) return true;
+ if(o==null||getClass()!=o.getClass()) return false;
+
+ TableQueryConfig that=(TableQueryConfig)o;
+
+ if(autoAdjustRanges!=that.autoAdjustRanges) return false;
+ if(columns!=null?!columns.equals(that.columns):that.columns!=null) return false;
+ if(iterators!=null?!iterators.equals(that.iterators):that.iterators!=null) return false;
+ if(ranges!=null?!ranges.equals(that.ranges):that.ranges!=null) return false;
+ if(tableName!=null?!tableName.equals(that.tableName):that.tableName!=null) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode(){
+ int result=tableName!=null?tableName.hashCode():0;
+ result=31*result+(iterators!=null?iterators.hashCode():0);
+ result=31*result+(ranges!=null?ranges.hashCode():0);
+ result=31*result+(columns!=null?columns.hashCode():0);
+ result=31*result+(autoAdjustRanges?1:0);
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/main/java/org/apache/accumulo/core/master/thrift/MasterClientService.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/master/thrift/MasterClientService.java b/core/src/main/java/org/apache/accumulo/core/master/thrift/MasterClientService.java
index 5b9949a..3a00518 100644
--- a/core/src/main/java/org/apache/accumulo/core/master/thrift/MasterClientService.java
+++ b/core/src/main/java/org/apache/accumulo/core/master/thrift/MasterClientService.java
@@ -1690,7 +1690,7 @@ import org.slf4j.LoggerFactory;
return TINFO;
case 1: // CREDENTIALS
return CREDENTIALS;
- case 2: // TABLE_NAME
+ case 2: // TABLE
return TABLE_NAME;
default:
return null;
@@ -2114,7 +2114,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 2: // TABLE_NAME
+ case 2: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
@@ -2824,7 +2824,7 @@ import org.slf4j.LoggerFactory;
return TINFO;
case 1: // CREDENTIALS
return CREDENTIALS;
- case 2: // TABLE_NAME
+ case 2: // TABLE
return TABLE_NAME;
case 6: // START_ROW
return START_ROW;
@@ -3562,7 +3562,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 2: // TABLE_NAME
+ case 2: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
@@ -4254,7 +4254,7 @@ import org.slf4j.LoggerFactory;
return TINFO;
case 1: // CREDENTIALS
return CREDENTIALS;
- case 2: // TABLE_NAME
+ case 2: // TABLE
return TABLE_NAME;
case 3: // PROPERTY
return PROPERTY;
@@ -4826,7 +4826,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 2: // TABLE_NAME
+ case 2: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
@@ -5473,7 +5473,7 @@ import org.slf4j.LoggerFactory;
return TINFO;
case 1: // CREDENTIALS
return CREDENTIALS;
- case 2: // TABLE_NAME
+ case 2: // TABLE
return TABLE_NAME;
case 3: // PROPERTY
return PROPERTY;
@@ -5971,7 +5971,7 @@ import org.slf4j.LoggerFactory;
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
- case 2: // TABLE_NAME
+ case 2: // TABLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.tableName = iprot.readString();
struct.setTableNameIsSet(true);
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/main/java/org/apache/accumulo/core/util/ArgumentChecker.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/util/ArgumentChecker.java b/core/src/main/java/org/apache/accumulo/core/util/ArgumentChecker.java
index 0c8ba07..2379873 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/ArgumentChecker.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/ArgumentChecker.java
@@ -60,4 +60,9 @@ public class ArgumentChecker {
if (i <= 0)
throw new IllegalArgumentException("integer should be > 0, was " + i);
}
+
+ public static final void notEmpty(Iterable arg) {
+ if (!arg.iterator().hasNext())
+ throw new IllegalArgumentException("Argument should not be empty");
+ }
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/a3f5cdc6/core/src/test/java/org/apache/accumulo/core/client/mapred/AccumuloInputFormatTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/mapred/AccumuloInputFormatTest.java b/core/src/test/java/org/apache/accumulo/core/client/mapred/AccumuloInputFormatTest.java
index 4f527e1..96113b1 100644
--- a/core/src/test/java/org/apache/accumulo/core/client/mapred/AccumuloInputFormatTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/client/mapred/AccumuloInputFormatTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.accumulo.core.client.mapred;
+import static org.apache.commons.codec.binary.Base64.encodeBase64;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -64,19 +65,18 @@ public class AccumuloInputFormatTest {
@Test
public void testSetIterator() throws IOException {
JobConf job = new JobConf();
-
IteratorSetting is = new IteratorSetting(1, "WholeRow", "org.apache.accumulo.core.iterators.WholeRowIterator");
AccumuloInputFormat.addIterator(job, is);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
is.write(new DataOutputStream(baos));
String iterators = job.get("AccumuloInputFormat.ScanOpts.Iterators");
- assertEquals(new String(Base64.encodeBase64(baos.toByteArray())), iterators);
+ assertEquals(new String(encodeBase64 (baos.toByteArray ())), iterators);
}
@Test
public void testAddIterator() throws IOException {
JobConf job = new JobConf();
-
+
AccumuloInputFormat.addIterator(job, new IteratorSetting(1, "WholeRow", WholeRowIterator.class));
AccumuloInputFormat.addIterator(job, new IteratorSetting(2, "Versions", "org.apache.accumulo.core.iterators.VersioningIterator"));
IteratorSetting iter = new IteratorSetting(3, "Count", "org.apache.accumulo.core.iterators.CountingIterator");
@@ -152,7 +152,7 @@ public class AccumuloInputFormatTest {
@Test
public void testGetIteratorSettings() throws IOException {
JobConf job = new JobConf();
-
+
AccumuloInputFormat.addIterator(job, new IteratorSetting(1, "WholeRow", "org.apache.accumulo.core.iterators.WholeRowIterator"));
AccumuloInputFormat.addIterator(job, new IteratorSetting(2, "Versions", "org.apache.accumulo.core.iterators.VersioningIterator"));
AccumuloInputFormat.addIterator(job, new IteratorSetting(3, "Count", "org.apache.accumulo.core.iterators.CountingIterator"));
@@ -185,7 +185,7 @@ public class AccumuloInputFormatTest {
JobConf job = new JobConf();
String regex = ">\"*%<>\'\\";
-
+
IteratorSetting is = new IteratorSetting(50, regex, RegExFilter.class);
RegExFilter.setRegexs(is, regex, null, null, null, false);
AccumuloInputFormat.addIterator(job, is);
[5/7] git commit: Removing deprecation for now until we have some
discussions. Updating/adding comments. ACCUMULO-391
Posted by cj...@apache.org.
Removing deprecation for now until we have some discussions. Updating/adding comments. ACCUMULO-391
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/793b5183
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/793b5183
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/793b5183
Branch: refs/heads/ACCUMULO-391
Commit: 793b5183b4caaa6b7da0cbbb53c26b4e1463eeaf
Parents: c8a8583
Author: Corey J. Nolet <cj...@gmail.com>
Authored: Sat Sep 28 23:23:48 2013 -0400
Committer: Corey J. Nolet <cj...@gmail.com>
Committed: Sat Sep 28 23:29:37 2013 -0400
----------------------------------------------------------------------
.../core/client/mapred/InputFormatBase.java | 2 +
.../core/client/mapreduce/InputFormatBase.java | 71 +++++++++++++-------
.../mapreduce/lib/util/InputConfigurator.java | 40 +++--------
.../mapreduce/AccumuloInputFormatTest.java | 15 ++---
4 files changed, 64 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/793b5183/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java b/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
index fb1b193..15c2dd2 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java
@@ -314,6 +314,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* the ranges that will be mapped over
* @since 1.5.0
*/
+ @Deprecated
public static void setRanges(JobConf job, Collection<Range> ranges) {
InputConfigurator.setRanges(CLASS, job, ranges);
}
@@ -329,6 +330,7 @@ public abstract class InputFormatBase<K,V> implements InputFormat<K,V> {
* @since 1.5.0
* @see #setRanges(JobConf, Collection)
*/
+ @Deprecated
protected static List<Range> getRanges(JobConf job) throws IOException {
return InputConfigurator.getRanges(CLASS,job);
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/793b5183/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
index d5208b4..5392459 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java
@@ -157,7 +157,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setConnectorInfo(Job, String, AuthenticationToken)
*/
protected static String getPrincipal(JobContext context) {
- return InputConfigurator.getPrincipal(CLASS, getConfiguration(context));
+ return InputConfigurator.getPrincipal(CLASS,getConfiguration(context));
}
/**
@@ -170,7 +170,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setInputTableName(Job, String)
*/
protected static String getInputTableName(JobContext context) {
- return InputConfigurator.getInputTableName(CLASS, getConfiguration(context));
+ return InputConfigurator.getInputTableName(CLASS,getConfiguration(context));
}
/**
@@ -206,7 +206,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setConnectorInfo(Job, String, String)
*/
protected static AuthenticationToken getAuthenticationToken(JobContext context) {
- return InputConfigurator.getAuthenticationToken(CLASS, getConfiguration(context));
+ return InputConfigurator.getAuthenticationToken(CLASS,getConfiguration(context));
}
/**
@@ -234,7 +234,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
*/
public static void setMockInstance(Job job, String instanceName) {
- InputConfigurator.setMockInstance(CLASS, job.getConfiguration(), instanceName);
+ InputConfigurator.setMockInstance(CLASS,job.getConfiguration(),instanceName);
}
/**
@@ -286,7 +286,6 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* the table to use when the tablename is null in the write call
* @since 1.5.0
*/
- @Deprecated
public static void setInputTableName(Job job, String tableName) {
InputConfigurator.setInputTableName(CLASS, job.getConfiguration(), tableName);
}
@@ -301,7 +300,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
*/
public static void setScanAuthorizations(Job job, Authorizations auths) {
- InputConfigurator.setScanAuthorizations(CLASS, job.getConfiguration(), auths);
+ InputConfigurator.setScanAuthorizations(CLASS,job.getConfiguration(),auths);
}
/**
@@ -327,7 +326,6 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* the ranges that will be mapped over
* @since 1.5.0
*/
- @Deprecated
public static void setRanges(Job job, Collection<Range> ranges) {
InputConfigurator.setRanges(CLASS, job.getConfiguration(), ranges);
}
@@ -342,7 +340,6 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setRanges(Job, Collection)
* @see #setRanges(org.apache.hadoop.mapreduce.Job, java.util.Collection)
*/
- @Deprecated
protected static List<Range> getRanges(JobContext context) throws IOException {
return InputConfigurator.getRanges(CLASS, getConfiguration(context));
}
@@ -357,7 +354,6 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* selected. An empty set is the default and is equivalent to scanning the all columns.
* @since 1.5.0
*/
- @Deprecated
public static void fetchColumns(Job job, Collection<Pair<Text,Text>> columnFamilyColumnQualifierPairs) {
InputConfigurator.fetchColumns(CLASS, job.getConfiguration(), columnFamilyColumnQualifierPairs);
}
@@ -384,14 +380,12 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* the configuration of the iterator
* @since 1.5.0
*/
- @Deprecated
public static void addIterator(Job job, IteratorSetting cfg) {
InputConfigurator.addIterator(CLASS, job.getConfiguration(), cfg);
}
/**
- * Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration. This will only return iterators that have not been set
- * for a specific table.
+ * Gets a list of the iterator settings (for iterators to apply to a scanner) from this configuration.
*
* @param context
* the Hadoop context for the configured job
@@ -399,7 +393,6 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
* @see #addIterator(Job, IteratorSetting)
*/
- @Deprecated
protected static List<IteratorSetting> getIterators(JobContext context) {
return InputConfigurator.getIterators(CLASS, getConfiguration(context));
}
@@ -418,7 +411,6 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @see #setRanges(Job, Collection)
* @since 1.5.0
*/
- @Deprecated
public static void setAutoAdjustRanges(Job job, boolean enableFeature) {
InputConfigurator.setAutoAdjustRanges(CLASS, job.getConfiguration(), enableFeature);
}
@@ -432,21 +424,52 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* @since 1.5.0
* @see #setAutoAdjustRanges(Job, boolean)
*/
- @Deprecated
protected static boolean getAutoAdjustRanges(JobContext context) {
return InputConfigurator.getAutoAdjustRanges(CLASS, getConfiguration(context));
}
-
- protected static void setTableQueryConfigurations(JobContext job, TableQueryConfig... configs) {
+
+ /**
+ * Sets the {@link TableQueryConfig} objects on the given Hadoop configuration
+ * @param job
+ * the Hadoop job instance to be configured
+ * @param configs
+ * the table query configs to be set on the configuration.
+ */
+ public static void setTableQueryConfigs(JobContext job, TableQueryConfig... configs) {
checkNotNull(configs);
InputConfigurator.setTableQueryConfigs(CLASS, getConfiguration(job), configs);
}
-
- public static List<TableQueryConfig> getTableQueryConfigurations(JobContext job) {
+
+ /**
+ * Fetches all {@link TableQueryConfig}s that have been set on the given Hadoop configuration.
+ *
+ * <p>
+ * Note this also returns the {@link TableQueryConfig} representing the table configurations set through the single
+ * table input methods ({@link #setInputTableName(org.apache.hadoop.mapreduce.Job, String)},
+ * {@link #setRanges(org.apache.hadoop.mapreduce.Job, java.util.Collection)},
+ * {@link #fetchColumns(org.apache.hadoop.mapreduce.Job, java.util.Collection)},
+ * {@link #addIterator(org.apache.hadoop.mapreduce.Job, org.apache.accumulo.core.client.IteratorSetting)}, etc...)
+ *
+ * @param job
+ * the Hadoop job instance to be configured
+ * @return
+ */
+ public static List<TableQueryConfig> getTableQueryConfigs(JobContext job) {
return InputConfigurator.getTableQueryConfigs(CLASS, getConfiguration(job));
}
-
- protected static TableQueryConfig getTableQueryConfiguration(JobContext job, String tableName) {
+
+ /**
+ * Fetches a {@link TableQueryConfig} that has been set on the configuration for a specific table.
+ *
+ * <p>
+ * null is returned in the event that the table doesn't exist.
+ * @param job
+ * the Hadoop job instance to be configured
+ * @param tableName
+ * the table name for which to grab the config object
+ * @return the {@link TableQueryConfig} for the given table
+ */
+ public static TableQueryConfig getTableQueryConfig(JobContext job, String tableName) {
return InputConfigurator.getTableQueryConfigs(CLASS, getConfiguration(job), tableName);
}
@@ -615,7 +638,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
* the table name for which to set up the iterators
*/
protected void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName) {
- TableQueryConfig config = getTableQueryConfiguration(context, tableName);
+ TableQueryConfig config = getTableQueryConfig(context,tableName);
List<IteratorSetting> iterators = config.getIterators();
for (IteratorSetting iterator : iterators)
scanner.addScanIterator(iterator);
@@ -633,7 +656,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
Instance instance = getInstance(attempt);
String principal = getPrincipal(attempt);
- TableQueryConfig tableConfig = getTableQueryConfiguration(attempt, split.getTableName());
+ TableQueryConfig tableConfig = getTableQueryConfig(attempt,split.getTableName());
// in case the table name changed, we can still use the previous name for terms of configuration,
// but for the scanner, we'll need to reference the new table name.
@@ -828,7 +851,7 @@ public abstract class InputFormatBase<K,V> extends InputFormat<K,V> {
validateOptions(conf);
LinkedList<InputSplit> splits = new LinkedList<InputSplit>();
- List<TableQueryConfig> tableConfigs = getTableQueryConfigurations(conf);
+ List<TableQueryConfig> tableConfigs = getTableQueryConfigs(conf);
for (TableQueryConfig tableConfig : tableConfigs) {
boolean autoAdjust = tableConfig.shouldAutoAdjustRanges();
http://git-wip-us.apache.org/repos/asf/accumulo/blob/793b5183/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
index 5e02177..82df3a1 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/util/InputConfigurator.java
@@ -89,9 +89,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @param tableName
* the table to use when the tablename is null in the write call
* @since 1.5.0
- * @deprecated since 1.6.0
*/
- @Deprecated
public static void setInputTableName(Class<?> implementingClass, Configuration conf, String tableName) {
notNull(tableName);
conf.set(enumToConfKey(implementingClass, ScanOpts.TABLE), tableName);
@@ -105,9 +103,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @param conf
* the Hadoop configuration object to configure
* @since 1.5.0
- * @deprecated since 1.6.0
*/
- @Deprecated
public static String getInputTableName(Class<?> implementingClass, Configuration conf) {
return conf.get(enumToConfKey(implementingClass, ScanOpts.TABLE));
}
@@ -156,9 +152,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @throws IllegalArgumentException
* if the ranges cannot be encoded into base 64
* @since 1.5.0
- * @deprecated since 1.6.0
*/
- @Deprecated
public static void setRanges(Class<?> implementingClass, Configuration conf, Collection<Range> ranges) {
notNull(ranges);
@@ -186,10 +180,8 @@ public class InputConfigurator extends ConfiguratorBase {
* @throws IOException
* if the ranges have been encoded improperly
* @since 1.6.0
- * @deprecated since 1.6.0
* @see #setRanges(Class, Configuration, Collection)
*/
- @Deprecated
public static List<Range> getRanges(Class<?> implementingClass, Configuration conf) throws IOException {
Collection<String> encodedRanges = conf.getStringCollection(enumToConfKey(implementingClass, ScanOpts.RANGES));
@@ -254,7 +246,6 @@ public class InputConfigurator extends ConfiguratorBase {
* @since 1.5.0
* @deprecated since 1.6.0
*/
- @Deprecated
public static void fetchColumns(Class<?> implementingClass, Configuration conf, Collection<Pair<Text,Text>> columnFamilyColumnQualifierPairs) {
notNull(columnFamilyColumnQualifierPairs);
ArrayList<String> columnStrings = new ArrayList<String>();
@@ -306,9 +297,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @throws IllegalArgumentException
* if the iterator can't be serialized into the configuration
* @since 1.5.0
- * @deprecated since 1.6.0
*/
- @Deprecated
public static void addIterator(Class<?> implementingClass, Configuration conf, IteratorSetting cfg) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
String newIter;
@@ -348,9 +337,7 @@ public class InputConfigurator extends ConfiguratorBase {
* the feature is enabled if true, disabled otherwise
* @see #setRanges(Class, Configuration, Collection)
* @since 1.5.0
- * @deprecated since 1.6.0
*/
- @Deprecated
public static void setAutoAdjustRanges(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
conf.setBoolean(enumToConfKey(implementingClass, Features.AUTO_ADJUST_RANGES), enableFeature);
}
@@ -365,9 +352,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @return false if the feature is disabled, true otherwise
* @since 1.5.0
* @see #setAutoAdjustRanges(Class, Configuration, boolean)
- * @deprecated since 1.6.0
*/
- @Deprecated
public static Boolean getAutoAdjustRanges(Class<?> implementingClass, Configuration conf) {
return conf.getBoolean(enumToConfKey(implementingClass, Features.AUTO_ADJUST_RANGES), true);
}
@@ -385,9 +370,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @param enableFeature
* the feature is enabled if true, disabled otherwise
* @since 1.5.0
- * @deprecated since 1.6.0
*/
- @Deprecated
public static void setScanIsolation(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
conf.setBoolean(enumToConfKey(implementingClass, Features.SCAN_ISOLATION), enableFeature);
}
@@ -402,9 +385,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @return true if the feature is enabled, false otherwise
* @since 1.5.0
* @see #setScanIsolation(Class, Configuration, boolean)
- * @deprecated since 1.6.0
*/
- @Deprecated
public static Boolean isIsolated(Class<?> implementingClass, Configuration conf) {
return conf.getBoolean(enumToConfKey(implementingClass, Features.SCAN_ISOLATION), false);
}
@@ -423,9 +404,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @param enableFeature
* the feature is enabled if true, disabled otherwise
* @since 1.5.0
- * @deprecated since 1.6.0
*/
- @Deprecated
public static void setLocalIterators(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
conf.setBoolean(enumToConfKey(implementingClass, Features.USE_LOCAL_ITERATORS), enableFeature);
}
@@ -440,9 +419,7 @@ public class InputConfigurator extends ConfiguratorBase {
* @return true if the feature is enabled, false otherwise
* @since 1.5.0
* @see #setLocalIterators(Class, Configuration, boolean)
- * @deprecated since 1.6.0
*/
- @Deprecated
public static Boolean usesLocalIterators(Class<?> implementingClass, Configuration conf) {
return conf.getBoolean(enumToConfKey(implementingClass, Features.USE_LOCAL_ITERATORS), false);
}
@@ -615,9 +592,6 @@ public class InputConfigurator extends ConfiguratorBase {
}
}
}
-
- // TODO: Check for the "default table case"
-
} catch (AccumuloException e) {
throw new IOException(e);
} catch (AccumuloSecurityException e) {
@@ -627,7 +601,17 @@ public class InputConfigurator extends ConfiguratorBase {
}
}
-
+ /**
+ * Returns the {@link TableQueryConfig} for the configuration based on the properties set using the single-table
+ * input methods.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop instance for which to retrieve the configuration
+ * @return the config object built from the single input table properties set on the job
+ * @throws IOException
+ */
protected static TableQueryConfig getDefaultTableConfig(Class<?> implementingClass, Configuration conf) throws IOException {
String tableName = getInputTableName(implementingClass, conf);
if(tableName != null) {
@@ -650,6 +634,4 @@ public class InputConfigurator extends ConfiguratorBase {
}
return null;
}
-
-
}
http://git-wip-us.apache.org/repos/asf/accumulo/blob/793b5183/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
index 1b4aa26..f9dd9ec 100644
--- a/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
@@ -25,15 +25,9 @@ import static org.junit.Assert.fail;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
-import java.util.Map;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
@@ -47,7 +41,6 @@ import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.user.RegExFilter;
import org.apache.accumulo.core.iterators.user.WholeRowIterator;
-import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.util.CachedConfiguration;
import org.apache.accumulo.core.util.Pair;
import org.apache.commons.codec.binary.Base64;
@@ -264,7 +257,7 @@ public class AccumuloInputFormatTest {
TableQueryConfig tableConfig1 = new TableQueryConfig(table1);
TableQueryConfig tableConfig2 = new TableQueryConfig(table2);
- AccumuloInputFormat.setTableQueryConfigurations(job, tableConfig1, tableConfig2);
+ AccumuloInputFormat.setTableQueryConfigs(job,tableConfig1,tableConfig2);
AccumuloInputFormat.setMockInstance(job, INSTANCE_NAME);
job.setMapperClass(TestMapper.class);
@@ -326,9 +319,9 @@ public class AccumuloInputFormatTest {
.setColumns(Collections.singleton(new Pair<Text,Text>(new Text("CF1"), new Text("CQ1"))))
.setIterators(Collections.singletonList(new IteratorSetting(50, "iter1", "iterclass1")));
- AccumuloInputFormat.setTableQueryConfigurations(job, table1, table2);
+ AccumuloInputFormat.setTableQueryConfigs(job,table1,table2);
- assertEquals(table1, AccumuloInputFormat.getTableQueryConfiguration(job, TEST_TABLE_1));
- assertEquals(table2, AccumuloInputFormat.getTableQueryConfiguration(job, TEST_TABLE_2));
+ assertEquals(table1, AccumuloInputFormat.getTableQueryConfig(job,TEST_TABLE_1));
+ assertEquals(table2, AccumuloInputFormat.getTableQueryConfig(job,TEST_TABLE_2));
}
}