http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/1d33b435/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/MergeToolMapper.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/MergeToolMapper.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/MergeToolMapper.java index 26a00bd..f5a9dae 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/MergeToolMapper.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/MergeToolMapper.java @@ -1,24 +1,22 @@ -package org.apache.rya.accumulo.mr.merge.mappers; - /* - * #%L - * org.apache.rya.accumulo.mr.merge - * %% - * Copyright (C) 2014 Rya - * %% - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * #L% + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. */ +package org.apache.rya.accumulo.mr.merge.mappers; import java.io.IOException; import java.util.Collection; @@ -160,7 +158,7 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { * @throws IOException */ @Override - public void run(Context context) throws IOException, InterruptedException { + public void run(final Context context) throws IOException, InterruptedException { setup(context); this.context = context; @@ -182,23 +180,23 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { parentRyaStatement = nextParentRyaStatement(); break; case ADVANCE_CHILD_AND_ADD: - RyaStatement tempChildRyaStatement = childRyaStatement; + final RyaStatement tempChildRyaStatement = childRyaStatement; childRyaStatement = nextChildRyaStatement(); addKey(tempChildRyaStatement, context); break; case ADVANCE_PARENT_AND_DELETE: - RyaStatement tempParentRyaStatement = parentRyaStatement; + final RyaStatement tempParentRyaStatement = parentRyaStatement; parentRyaStatement = nextParentRyaStatement(); deleteKey(tempParentRyaStatement, context); break; case ADVANCE_BOTH: - ColumnVisibility cv1 = new ColumnVisibility(parentRyaStatement.getColumnVisibility()); - ColumnVisibility cv2 = new ColumnVisibility(childRyaStatement.getColumnVisibility()); + final ColumnVisibility cv1 = new ColumnVisibility(parentRyaStatement.getColumnVisibility()); + final ColumnVisibility cv2 = new ColumnVisibility(childRyaStatement.getColumnVisibility()); // Update new column visibility now if necessary if (!cv1.equals(cv2) && !cv2.equals(AccumuloRdfConstants.EMPTY_CV)) { - ColumnVisibility newCv = combineColumnVisibilities(cv1, cv2); - RyaStatement newCvRyaStatement = updateRyaStatementColumnVisibility(parentRyaStatement, newCv); + final ColumnVisibility newCv = combineColumnVisibilities(cv1, cv2); + final RyaStatement newCvRyaStatement = updateRyaStatementColumnVisibility(parentRyaStatement, newCv); deleteKey(parentRyaStatement, context); addKey(newCvRyaStatement, context); @@ -230,50 +228,50 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { return nextRyaStatement(childIterator, childRyaContext); } - private static RyaStatement nextRyaStatement(Iterator<Entry<Key, Value>> iterator, RyaTripleContext ryaContext) { + private static RyaStatement nextRyaStatement(final Iterator<Entry<Key, Value>> iterator, final RyaTripleContext ryaContext) { RyaStatement ryaStatement = null; if (iterator.hasNext()) { - Entry<Key, Value> entry = iterator.next(); - Key key = entry.getKey(); - Value value = entry.getValue(); + final Entry<Key, Value> entry = iterator.next(); + final Key key = entry.getKey(); + final Value value = entry.getValue(); try { ryaStatement = createRyaStatement(key, value, ryaContext); - } catch (TripleRowResolverException e) { + } catch (final TripleRowResolverException e) { log.error("TripleRowResolverException encountered while creating statement", e); } } return ryaStatement; } - private static RyaStatement nextRyaStatement(Context context, RyaTripleContext ryaContext) throws IOException, InterruptedException { + private static RyaStatement nextRyaStatement(final Context context, final RyaTripleContext ryaContext) throws IOException, InterruptedException { RyaStatement ryaStatement = null; if (context.nextKeyValue()) { - Key key = context.getCurrentKey(); - Value value = context.getCurrentValue(); + final Key key = context.getCurrentKey(); + final Value value = context.getCurrentValue(); try { ryaStatement = createRyaStatement(key, value, ryaContext); - } catch (TripleRowResolverException e) { + } catch (final TripleRowResolverException e) { log.error("TripleRowResolverException encountered while creating statement", e); } } return ryaStatement; } - private static RyaStatement createRyaStatement(Key key, Value value, RyaTripleContext ryaTripleContext) throws TripleRowResolverException { - byte[] row = key.getRowData() != null && key.getRowData().toArray().length > 0 ? key.getRowData().toArray() : null; - byte[] columnFamily = key.getColumnFamilyData() != null && key.getColumnFamilyData().toArray().length > 0 ? key.getColumnFamilyData().toArray() : null; - byte[] columnQualifier = key.getColumnQualifierData() != null && key.getColumnQualifierData().toArray().length > 0 ? key.getColumnQualifierData().toArray() : null; - Long timestamp = key.getTimestamp(); - byte[] columnVisibility = key.getColumnVisibilityData() != null && key.getColumnVisibilityData().toArray().length > 0 ? key.getColumnVisibilityData().toArray() : null; - byte[] valueBytes = value != null && value.get().length > 0 ? value.get() : null; - TripleRow tripleRow = new TripleRow(row, columnFamily, columnQualifier, timestamp, columnVisibility, valueBytes); - RyaStatement ryaStatement = ryaTripleContext.deserializeTriple(TABLE_LAYOUT.SPO, tripleRow); + private static RyaStatement createRyaStatement(final Key key, final Value value, final RyaTripleContext ryaTripleContext) throws TripleRowResolverException { + final byte[] row = key.getRowData() != null && key.getRowData().toArray().length > 0 ? key.getRowData().toArray() : null; + final byte[] columnFamily = key.getColumnFamilyData() != null && key.getColumnFamilyData().toArray().length > 0 ? key.getColumnFamilyData().toArray() : null; + final byte[] columnQualifier = key.getColumnQualifierData() != null && key.getColumnQualifierData().toArray().length > 0 ? key.getColumnQualifierData().toArray() : null; + final Long timestamp = key.getTimestamp(); + final byte[] columnVisibility = key.getColumnVisibilityData() != null && key.getColumnVisibilityData().toArray().length > 0 ? key.getColumnVisibilityData().toArray() : null; + final byte[] valueBytes = value != null && value.get().length > 0 ? value.get() : null; + final TripleRow tripleRow = new TripleRow(row, columnFamily, columnQualifier, timestamp, columnVisibility, valueBytes); + final RyaStatement ryaStatement = ryaTripleContext.deserializeTriple(TABLE_LAYOUT.SPO, tripleRow); return ryaStatement; } @Override - protected void setup(Context context) throws IOException, InterruptedException { + protected void setup(final Context context) throws IOException, InterruptedException { super.setup(context); log.info("Setting up mapper"); @@ -323,14 +321,14 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { // Find the parent's time offset that was stored when the child was copied. parentTimeOffset = AccumuloRyaUtils.getTimeOffset(childDao); - String durationBreakdown = TimeUtils.getDurationBreakdown(parentTimeOffset); + final String durationBreakdown = TimeUtils.getDurationBreakdown(parentTimeOffset); log.info("The table " + parentTableName + " has a time offset of: " + durationBreakdown); childTimeOffset = Long.valueOf(childConfig.get(CopyTool.CHILD_TIME_OFFSET_PROP, null)); - Date adjustedParentStartTime = new Date(startTime.getTime() - parentTimeOffset); - Date adjustedChildStartTime = new Date(startTime.getTime() - childTimeOffset); + final Date adjustedParentStartTime = new Date(startTime.getTime() - parentTimeOffset); + final Date adjustedChildStartTime = new Date(startTime.getTime() - childTimeOffset); log.info("Adjusted parent start time: " + adjustedParentStartTime); log.info("Adjusted child start time: " + adjustedChildStartTime); - } catch (RyaDAOException e) { + } catch (final RyaDAOException e) { log.error("Error getting time offset", e); } } @@ -345,8 +343,8 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { * @return the new {@link Configuration} where the parent connection values are replaced with * the child connection values. */ - public static Configuration getChildConfig(Configuration parentConfig) { - Configuration childConfig = new Configuration(parentConfig); + public static Configuration getChildConfig(final Configuration parentConfig) { + final Configuration childConfig = new Configuration(parentConfig); // Switch the temp child properties to be the main ones convertChildPropToParentProp(childConfig, parentConfig, MRUtils.AC_MOCK_PROP); @@ -370,8 +368,8 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { * @param parentConfig the parent/main {@link Configuration}. * @param parentPropertyName the parent property name. */ - public static void convertChildPropToParentProp(Configuration childConfig, Configuration parentConfig, String parentPropertyName) { - String childValue = parentConfig.get(parentPropertyName + MergeTool.CHILD_SUFFIX, ""); + public static void convertChildPropToParentProp(final Configuration childConfig, final Configuration parentConfig, final String parentPropertyName) { + final String childValue = parentConfig.get(parentPropertyName + MergeTool.CHILD_SUFFIX, ""); childConfig.set(parentPropertyName, childValue); } @@ -381,7 +379,7 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { * @param cv2 the second (child) {@link ColumnVisibility}. * @return the newly combined {@link ColumnVisibility}. */ - public static ColumnVisibility combineColumnVisibilities(ColumnVisibility cv1, ColumnVisibility cv2) { + public static ColumnVisibility combineColumnVisibilities(final ColumnVisibility cv1, final ColumnVisibility cv2) { // OR the 2 column visibilities together if they're different String columnVisibilityExpression; if (cv1.equals(AccumuloRdfConstants.EMPTY_CV)) { @@ -395,45 +393,45 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { return newCv; } - private Scanner setupChildScanner(Context context) throws IOException { + private Scanner setupChildScanner(final Context context) throws IOException { return setupScanner(context, childTableName, childConfig); } - private static Scanner setupScanner(Context context, String tableName, Configuration config) throws IOException { - RangeInputSplit split = (RangeInputSplit) context.getInputSplit(); - Range splitRange = split.getRange(); - Scanner scanner = AccumuloRyaUtils.getScanner(tableName, config); + private static Scanner setupScanner(final Context context, final String tableName, final Configuration config) throws IOException { + final RangeInputSplit split = (RangeInputSplit) context.getInputSplit(); + final Range splitRange = split.getRange(); + final Scanner scanner = AccumuloRyaUtils.getScanner(tableName, config); scanner.setRange(splitRange); return scanner; } - private void writeRyaMutations(RyaStatement ryaStatement, Context context, boolean isDelete) throws IOException, InterruptedException { + private void writeRyaMutations(final RyaStatement ryaStatement, final Context context, final boolean isDelete) throws IOException, InterruptedException { if (ryaStatement.getColumnVisibility() == null) { ryaStatement.setColumnVisibility(AccumuloRdfConstants.EMPTY_CV.getExpression()); } - Map<TABLE_LAYOUT, Collection<Mutation>> mutationMap = ryaTableMutationFactory.serialize(ryaStatement); - Collection<Mutation> spoMutations = mutationMap.get(TABLE_LAYOUT.SPO); - Collection<Mutation> poMutations = mutationMap.get(TABLE_LAYOUT.PO); - Collection<Mutation> ospMutations = mutationMap.get(TABLE_LAYOUT.OSP); + final Map<TABLE_LAYOUT, Collection<Mutation>> mutationMap = ryaTableMutationFactory.serialize(ryaStatement); + final Collection<Mutation> spoMutations = mutationMap.get(TABLE_LAYOUT.SPO); + final Collection<Mutation> poMutations = mutationMap.get(TABLE_LAYOUT.PO); + final Collection<Mutation> ospMutations = mutationMap.get(TABLE_LAYOUT.OSP); - for (Mutation mutation : spoMutations) { + for (final Mutation mutation : spoMutations) { writeMutation(spoTable, mutation, context, isDelete); } - for (Mutation mutation : poMutations) { + for (final Mutation mutation : poMutations) { writeMutation(poTable, mutation, context, isDelete); } - for (Mutation mutation : ospMutations) { + for (final Mutation mutation : ospMutations) { writeMutation(ospTable, mutation, context, isDelete); } } - private void addKey(RyaStatement ryaStatement, Context context) throws IOException, InterruptedException { + private void addKey(final RyaStatement ryaStatement, final Context context) throws IOException, InterruptedException { writeRyaMutations(ryaStatement, context, false); } - private void deleteKey(RyaStatement ryaStatement, Context context) throws IOException, InterruptedException { + private void deleteKey(final RyaStatement ryaStatement, final Context context) throws IOException, InterruptedException { writeRyaMutations(ryaStatement, context, true); } @@ -447,12 +445,12 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { * @throws IOException * @throws InterruptedException */ - private static void writeMutation(Text table, Mutation mutation, Context context, boolean isDelete) throws IOException, InterruptedException { + private static void writeMutation(final Text table, final Mutation mutation, final Context context, final boolean isDelete) throws IOException, InterruptedException { if (isDelete) { - List<ColumnUpdate> updates = mutation.getUpdates(); - ColumnUpdate columnUpdate = updates.get(0); - ColumnVisibility cv = columnUpdate.getColumnVisibility() != null ? new ColumnVisibility(columnUpdate.getColumnVisibility()) : null; - Mutation deleteMutation = new Mutation(new Text(mutation.getRow())); + final List<ColumnUpdate> updates = mutation.getUpdates(); + final ColumnUpdate columnUpdate = updates.get(0); + final ColumnVisibility cv = columnUpdate.getColumnVisibility() != null ? new ColumnVisibility(columnUpdate.getColumnVisibility()) : null; + final Mutation deleteMutation = new Mutation(new Text(mutation.getRow())); deleteMutation.putDelete(columnUpdate.getColumnFamily(), columnUpdate.getColumnQualifier(), cv, columnUpdate.getTimestamp()); context.write(table, deleteMutation); } else { @@ -467,7 +465,7 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { * {@code false} if it's from the child instance. * @return the normalized {@link Date} or the same date if nothing needed to be adjusted. */ - private Date normalizeDate(Date date, boolean isParentTable) { + private Date normalizeDate(final Date date, final boolean isParentTable) { Date normalizedDate = date; if (useTimeSync) { if (isParentTable) { @@ -502,7 +500,7 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { * @throws InterruptedException * @throws TripleRowResolverException */ - private CompareKeysResult compareKeys(RyaStatement key1, RyaStatement key2) throws MutationsRejectedException, IOException, InterruptedException, TripleRowResolverException { + private CompareKeysResult compareKeys(final RyaStatement key1, final RyaStatement key2) throws MutationsRejectedException, IOException, InterruptedException, TripleRowResolverException { log.trace("key1 = " + key1); log.trace("key2 = " + key2); if (key1 == null && key2 == null) { @@ -510,38 +508,38 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { return CompareKeysResult.FINISHED; } else if (key1 == null) { // Reached the end of the parent table so add the remaining child keys if they meet the time criteria. - Date t2 = normalizeDate(new Date(key2.getTimestamp()), false); + final Date t2 = normalizeDate(new Date(key2.getTimestamp()), false); // Move on to next comparison (do nothing) or add this child key to parent - boolean doNothing = usesStartTime && t2.before(startTime); + final boolean doNothing = usesStartTime && t2.before(startTime); return doNothing ? CompareKeysResult.ADVANCE_CHILD : CompareKeysResult.ADVANCE_CHILD_AND_ADD; } else if (key2 == null) { // Reached the end of the child table so delete the remaining parent keys if they meet the time criteria. - Date t1 = normalizeDate(new Date(key1.getTimestamp()), true); + final Date t1 = normalizeDate(new Date(key1.getTimestamp()), true); // Move on to next comparison (do nothing) or delete this key from parent - boolean doNothing = usesStartTime && (copyToolInputTime != null && (t1.before(copyToolInputTime) || (t1.after(copyToolInputTime) && t1.after(startTime))) || (copyToolInputTime == null && t1.after(startTime))); + final boolean doNothing = usesStartTime && (copyToolInputTime != null && (t1.before(copyToolInputTime) || (t1.after(copyToolInputTime) && t1.after(startTime))) || (copyToolInputTime == null && t1.after(startTime))); return doNothing ? CompareKeysResult.ADVANCE_PARENT : CompareKeysResult.ADVANCE_PARENT_AND_DELETE; } else { // There are 2 keys to compare - Map<TABLE_LAYOUT, TripleRow> map1 = parentRyaContext.serializeTriple(key1); - Text row1 = new Text(map1.get(TABLE_LAYOUT.SPO).getRow()); - Map<TABLE_LAYOUT, TripleRow> map2 = childRyaContext.serializeTriple(key2); - Text row2 = new Text(map2.get(TABLE_LAYOUT.SPO).getRow()); - Date t1 = normalizeDate(new Date(key1.getTimestamp()), true); - Date t2 = normalizeDate(new Date(key2.getTimestamp()), false); + final Map<TABLE_LAYOUT, TripleRow> map1 = parentRyaContext.serializeTriple(key1); + final Text row1 = new Text(map1.get(TABLE_LAYOUT.SPO).getRow()); + final Map<TABLE_LAYOUT, TripleRow> map2 = childRyaContext.serializeTriple(key2); + final Text row2 = new Text(map2.get(TABLE_LAYOUT.SPO).getRow()); + final Date t1 = normalizeDate(new Date(key1.getTimestamp()), true); + final Date t2 = normalizeDate(new Date(key2.getTimestamp()), false); if (row1.compareTo(row2) < 0) { // Parent key sort order was before the child key sort order // so it doesn't exist in the child table. // What does this mean? Was it added by the parent after the child was cloned? (Meaning we should leave it) // Or did the child delete it after it was cloned? (Meaning we should delete it) - boolean doNothing = usesStartTime && (copyToolInputTime != null && (t1.before(copyToolInputTime) || (t1.after(copyToolInputTime) && t1.after(startTime))) || (copyToolInputTime == null && t1.after(startTime))); + final boolean doNothing = usesStartTime && (copyToolInputTime != null && (t1.before(copyToolInputTime) || (t1.after(copyToolInputTime) && t1.after(startTime))) || (copyToolInputTime == null && t1.after(startTime))); return doNothing ? CompareKeysResult.ADVANCE_PARENT : CompareKeysResult.ADVANCE_PARENT_AND_DELETE; } else if (row1.compareTo(row2) > 0) { // Parent key sort order was after the child key sort order // so it doesn't exist in the parent table. // What does this mean? Was it deleted by the parent after the child was cloned? (Meaning we should leave it) // Or did the child add it after it was cloned? (Meaning we should add it) - boolean doNothing = usesStartTime && t2.before(startTime); + final boolean doNothing = usesStartTime && t2.before(startTime); return doNothing ? CompareKeysResult.ADVANCE_CHILD : CompareKeysResult.ADVANCE_CHILD_AND_ADD; } else { // Rows are the same. So just check if column visibility needs to be updated and @@ -551,13 +549,13 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { } } - private static RyaStatement updateRyaStatementColumnVisibility(RyaStatement ryaStatement, ColumnVisibility newCv) { - RyaStatement newCvRyaStatement = new RyaStatement(ryaStatement.getSubject(), ryaStatement.getPredicate(), ryaStatement.getObject(), ryaStatement.getContext(), ryaStatement.getQualifer(), newCv.getExpression(), ryaStatement.getValue(), ryaStatement.getTimestamp()); + private static RyaStatement updateRyaStatementColumnVisibility(final RyaStatement ryaStatement, final ColumnVisibility newCv) { + final RyaStatement newCvRyaStatement = new RyaStatement(ryaStatement.getSubject(), ryaStatement.getPredicate(), ryaStatement.getObject(), ryaStatement.getContext(), ryaStatement.getQualifer(), newCv.getExpression(), ryaStatement.getValue(), ryaStatement.getTimestamp()); return newCvRyaStatement; } @Override - protected void cleanup(Context context) throws IOException, InterruptedException { + protected void cleanup(final Context context) throws IOException, InterruptedException { super.cleanup(context); log.info("Cleaning up mapper..."); if (childScanner != null) { @@ -567,7 +565,7 @@ public class MergeToolMapper extends Mapper<Key, Value, Text, Mutation> { if (childDao != null) { childDao.destroy(); } - } catch (RyaDAOException e) { + } catch (final RyaDAOException e) { log.error("Error destroying child DAO", e); } log.info("Cleaned up mapper");
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/1d33b435/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/RowRuleMapper.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/RowRuleMapper.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/RowRuleMapper.java index c34e2f2..e224a07 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/RowRuleMapper.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/mappers/RowRuleMapper.java @@ -1,24 +1,22 @@ -package org.apache.rya.accumulo.mr.merge.mappers; - /* - * #%L - * org.apache.rya.accumulo.mr.merge - * %% - * Copyright (C) 2014 Rya - * %% - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * #L% + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. */ +package org.apache.rya.accumulo.mr.merge.mappers; import java.io.IOException; import java.util.Map; @@ -62,13 +60,13 @@ public class RowRuleMapper extends BaseRuleMapper<GroupedRow, GroupedRow> { public static final int MAX_STATEMENTS_DEFAULT = 10000; private static final Logger log = Logger.getLogger(RowRuleMapper.class); - private GroupedRow compositeKey = new GroupedRow(); - private GroupedRow compositeVal = new GroupedRow(); + private final GroupedRow compositeKey = new GroupedRow(); + private final GroupedRow compositeVal = new GroupedRow(); private int cachedStatements = 0; private int maxStatements; @Override - protected void setup(Context context) throws IOException, InterruptedException { + protected void setup(final Context context) throws IOException, InterruptedException { super.setup(context); // Set up a mock child DAO for caching serialized statements childAccumuloRdfConfiguration.setBoolean(MRUtils.AC_MOCK_PROP, true); @@ -85,7 +83,7 @@ public class RowRuleMapper extends BaseRuleMapper<GroupedRow, GroupedRow> { } @Override - protected void cleanup(Context context) throws IOException, InterruptedException { + protected void cleanup(final Context context) throws IOException, InterruptedException { flush(context); super.cleanup(context); } @@ -100,11 +98,11 @@ public class RowRuleMapper extends BaseRuleMapper<GroupedRow, GroupedRow> { * if Hadoop reports an error writing the in-memory tables to the output. */ @Override - protected void copyStatement(RyaStatement rstmt, Context context) throws IOException, InterruptedException { + protected void copyStatement(final RyaStatement rstmt, final Context context) throws IOException, InterruptedException { try { childDao.add(rstmt); cachedStatements++; - } catch (RyaDAOException e) { + } catch (final RyaDAOException e) { throw new IOException("Error serializing RyaStatement", e); } if (cachedStatements >= maxStatements) { @@ -121,7 +119,7 @@ public class RowRuleMapper extends BaseRuleMapper<GroupedRow, GroupedRow> { * @throws IOException If Hadoop reports an error writing the output */ @Override - protected void copyRow(Key key, Value value, Context context) throws IOException, InterruptedException { + protected void copyRow(final Key key, final Value value, final Context context) throws IOException, InterruptedException { compositeKey.setGroup(childTableName); compositeKey.setKey(key); compositeVal.setKey(key); @@ -133,19 +131,19 @@ public class RowRuleMapper extends BaseRuleMapper<GroupedRow, GroupedRow> { * Insert copy tool metadata, if the in-memory instance has been configured. */ @Override - protected void addMetadataKeys(Context context) throws IOException { + protected void addMetadataKeys(final Context context) throws IOException { try { if (childDao != null && childDao.isInitialized()) { if (runTime != null) { - RyaStatement ryaStatement = AccumuloRyaUtils.createCopyToolRunTimeRyaStatement(runTime); + final RyaStatement ryaStatement = AccumuloRyaUtils.createCopyToolRunTimeRyaStatement(runTime); copyStatement(ryaStatement, context); } if (startTime != null) { - RyaStatement ryaStatement = AccumuloRyaUtils.createCopyToolSplitTimeRyaStatement(startTime); + final RyaStatement ryaStatement = AccumuloRyaUtils.createCopyToolSplitTimeRyaStatement(startTime); copyStatement(ryaStatement, context); } if (timeOffset != null) { - RyaStatement ryaStatement = AccumuloRyaUtils.createTimeOffsetRyaStatement(timeOffset); + final RyaStatement ryaStatement = AccumuloRyaUtils.createTimeOffsetRyaStatement(timeOffset); copyStatement(ryaStatement, context); } } @@ -154,21 +152,21 @@ public class RowRuleMapper extends BaseRuleMapper<GroupedRow, GroupedRow> { } } - private void flush(Context context) throws IOException, InterruptedException { + private void flush(final Context context) throws IOException, InterruptedException { try { childDao.flush(); - } catch (RyaDAOException e) { + } catch (final RyaDAOException e) { throw new IOException("Error writing to in-memory table", e); } - TableOperations ops = childConnector.tableOperations(); - SecurityOperations secOps = childConnector.securityOperations(); + final TableOperations ops = childConnector.tableOperations(); + final SecurityOperations secOps = childConnector.securityOperations(); Authorizations childAuths; try { childAuths = secOps.getUserAuthorizations(childUser); } catch (AccumuloException | AccumuloSecurityException e) { throw new IOException("Error connecting to mock instance", e); } - for (String table : ops.list()) { + for (final String table : ops.list()) { // Only copy Rya tables (skip system tables) if (!table.startsWith(childTablePrefix)) { continue; @@ -177,8 +175,8 @@ public class RowRuleMapper extends BaseRuleMapper<GroupedRow, GroupedRow> { try { // Output every row in this mock table int rows = 0; - Scanner scanner = childDao.getConnector().createScanner(table, childAuths); - for (Map.Entry<Key, Value> row : scanner) { + final Scanner scanner = childDao.getConnector().createScanner(table, childAuths); + for (final Map.Entry<Key, Value> row : scanner) { compositeKey.setKey(row.getKey()); compositeVal.setKey(row.getKey()); compositeVal.setValue(row.getValue()); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/1d33b435/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/reducers/MultipleFileReducer.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/reducers/MultipleFileReducer.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/reducers/MultipleFileReducer.java index 7705f3a..162651a 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/reducers/MultipleFileReducer.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/reducers/MultipleFileReducer.java @@ -1,24 +1,22 @@ -package org.apache.rya.accumulo.mr.merge.reducers; - /* - * #%L - * org.apache.rya.accumulo.mr.merge - * %% - * Copyright (C) 2014 Rya - * %% - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * #L% + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. */ +package org.apache.rya.accumulo.mr.merge.reducers; import java.io.IOException; @@ -36,12 +34,12 @@ public class MultipleFileReducer extends Reducer<GroupedRow, GroupedRow, Key, Va private MultipleOutputs<Key, Value> mos; @Override - protected void setup(Context context) { + protected void setup(final Context context) { mos = new MultipleOutputs<>(context); } @Override - protected void cleanup(Context context) throws IOException, InterruptedException { + protected void cleanup(final Context context) throws IOException, InterruptedException { if (mos != null) { mos.close(); } @@ -54,10 +52,10 @@ public class MultipleFileReducer extends Reducer<GroupedRow, GroupedRow, Key, Va * @param context Context for writing */ @Override - protected void reduce(GroupedRow group, Iterable<GroupedRow> rows, Context context) throws IOException, InterruptedException { - String groupName = group.getGroup().toString(); - String destination = groupName + "/files/part"; - for (GroupedRow row : rows) { + protected void reduce(final GroupedRow group, final Iterable<GroupedRow> rows, final Context context) throws IOException, InterruptedException { + final String groupName = group.getGroup().toString(); + final String destination = groupName + "/files/part"; + for (final GroupedRow row : rows) { mos.write(row.getKey(), row.getValue(), destination); } } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/1d33b435/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java index 781f885..9d67e38 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloInstanceDriver.java @@ -1,24 +1,22 @@ -package org.apache.rya.accumulo.mr.merge.util; - /* - * #%L - * org.apache.rya.accumulo.mr.merge - * %% - * Copyright (C) 2014 Rya - * %% - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * #L% + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. */ +package org.apache.rya.accumulo.mr.merge.util; import java.io.File; import java.io.IOException; @@ -55,7 +53,6 @@ import org.apache.rya.accumulo.mr.MRUtils; import org.apache.rya.accumulo.mr.merge.MergeTool; import org.apache.rya.api.RdfCloudTripleStoreConstants; import org.apache.rya.api.persist.RyaDAOException; -import org.apache.rya.indexing.accumulo.ConfigUtils; import twitter4j.Logger; /** @@ -68,17 +65,17 @@ public class AccumuloInstanceDriver { public static final String ROOT_USER_NAME = "root"; - private String driverName; - private boolean isMock; - private boolean shouldCreateIndices; - private boolean isReadOnly; - private boolean isParent; + private final String driverName; + private final boolean isMock; + private final boolean shouldCreateIndices; + private final boolean isReadOnly; + private final boolean isParent; - private String user; - private String password; - private String instanceName; - private String tablePrefix; - private String auth; + private final String user; + private final String password; + private final String instanceName; + private final String tablePrefix; + private final String auth; private Connector connector; @@ -86,7 +83,7 @@ public class AccumuloInstanceDriver { private SecurityOperations secOps; - private AccumuloRdfConfiguration config = new AccumuloRdfConfiguration(); + private final AccumuloRdfConfiguration config = new AccumuloRdfConfiguration(); private MiniAccumuloCluster miniAccumuloCluster = null; @@ -98,11 +95,11 @@ public class AccumuloInstanceDriver { private String zooKeepers; - private Map<String, String> configMap = new LinkedHashMap<>(); + private final Map<String, String> configMap = new LinkedHashMap<>(); private List<String> indices = null; - private List<String> tableList = new ArrayList<>(); + private final List<String> tableList = new ArrayList<>(); private File tempDir = null; @@ -134,7 +131,7 @@ public class AccumuloInstanceDriver { * @param tablePrefix the table prefix. * @param auth the comma-separated authorization list. */ - public AccumuloInstanceDriver(String driverName, boolean isMock, boolean shouldCreateIndices, boolean isReadOnly, boolean isParent, String user, String password, String instanceName, String tablePrefix, String auth) { + public AccumuloInstanceDriver(final String driverName, final boolean isMock, final boolean shouldCreateIndices, final boolean isReadOnly, final boolean isParent, final String user, final String password, final String instanceName, final String tablePrefix, final String auth) { this.driverName = Preconditions.checkNotNull(driverName); this.isMock = isMock; this.shouldCreateIndices = shouldCreateIndices; @@ -200,11 +197,11 @@ public class AccumuloInstanceDriver { */ private void copyHadoopHomeToTemp() throws IOException { if (IS_COPY_HADOOP_HOME_ENABLED && SystemUtils.IS_OS_WINDOWS) { - String hadoopHomeEnv = System.getenv("HADOOP_HOME"); + final String hadoopHomeEnv = System.getenv("HADOOP_HOME"); if (hadoopHomeEnv != null) { - File hadoopHomeDir = new File(hadoopHomeEnv); + final File hadoopHomeDir = new File(hadoopHomeEnv); if (hadoopHomeDir.exists()) { - File binDir = Paths.get(hadoopHomeDir.getAbsolutePath(), "/bin").toFile(); + final File binDir = Paths.get(hadoopHomeDir.getAbsolutePath(), "/bin").toFile(); if (binDir.exists()) { FileUtils.copyDirectoryToDirectory(binDir, tempDir); } else { @@ -226,8 +223,8 @@ public class AccumuloInstanceDriver { public void setUpTables() throws Exception { // Setup tables and permissions log.info("Setting up " + driverName + " tables and permissions"); - for (String tableSuffix : TABLE_NAME_SUFFIXES) { - String tableName = tablePrefix + tableSuffix; + for (final String tableSuffix : TABLE_NAME_SUFFIXES) { + final String tableName = tablePrefix + tableSuffix; tableList.add(tableName); if (!connector.tableOperations().exists(tableName)) { connector.tableOperations().create(tableName); @@ -248,7 +245,7 @@ public class AccumuloInstanceDriver { tableList.addAll(indices); log.info("Setting up " + driverName + " indices"); - for (String index : indices) { + for (final String index : indices) { if (!connector.tableOperations().exists(index)) { connector.tableOperations().create(index); } @@ -262,12 +259,12 @@ public class AccumuloInstanceDriver { secOps.createLocalUser(user, new PasswordToken(password)); } addAuths(auth); - TablePermission tablePermission = isReadOnly ? TablePermission.READ : TablePermission.WRITE; - for (String tableSuffix : TABLE_NAME_SUFFIXES) { + final TablePermission tablePermission = isReadOnly ? TablePermission.READ : TablePermission.WRITE; + for (final String tableSuffix : TABLE_NAME_SUFFIXES) { secOps.grantTablePermission(user, tablePrefix + tableSuffix, tablePermission); } if (shouldCreateIndices) { - for (String index : indices) { + for (final String index : indices) { secOps.grantTablePermission(user, index, tablePermission); } } @@ -285,7 +282,7 @@ public class AccumuloInstanceDriver { dao.setConf(config); // Flush the tables before initializing the DAO - for (String tableName : tableList) { + for (final String tableName : tableList) { connector.tableOperations().flush(tableName, null, null, false); } @@ -311,10 +308,10 @@ public class AccumuloInstanceDriver { log.info(driverName + " config properties"); config.setTablePrefix(tablePrefix); - for (Entry<String, String> entry : configMap.entrySet()) { - String key = entry.getKey(); - String value = entry.getValue(); - String argument = ToolConfigUtils.makeArgument(isParent ? key : key + MergeTool.CHILD_SUFFIX, value); + for (final Entry<String, String> entry : configMap.entrySet()) { + final String key = entry.getKey(); + final String value = entry.getValue(); + final String argument = ToolConfigUtils.makeArgument(isParent ? key : key + MergeTool.CHILD_SUFFIX, value); log.info(argument); config.set(key, value); } @@ -329,7 +326,7 @@ public class AccumuloInstanceDriver { public void tearDownTables() throws Exception { // delete all tables. if (connector != null) { - for (String tableName : tableList) { + for (final String tableName : tableList) { if (connector.tableOperations().exists(tableName)) { connector.tableOperations().delete(tableName); } @@ -346,7 +343,7 @@ public class AccumuloInstanceDriver { log.info("Stopping " + driverName + " DAO"); try { dao.destroy(); - } catch (RyaDAOException e) { + } catch (final RyaDAOException e) { log.error("Error stopping " + driverName + " DAO", e); } dao = null; @@ -390,7 +387,7 @@ public class AccumuloInstanceDriver { if (tempDir != null) { try { FileUtils.deleteDirectory(tempDir); - } catch (IOException e) { + } catch (final IOException e) { log.error("Error deleting " + driverName + " temp directory", e); } tempDir = null; @@ -403,8 +400,8 @@ public class AccumuloInstanceDriver { * @throws AccumuloException * @throws AccumuloSecurityException */ - public void addAuths(String... auths) throws AccumuloException, AccumuloSecurityException { - Authorizations newAuths = AccumuloRyaUtils.addUserAuths(user, secOps, auths); + public void addAuths(final String... auths) throws AccumuloException, AccumuloSecurityException { + final Authorizations newAuths = AccumuloRyaUtils.addUserAuths(user, secOps, auths); secOps.changeUserAuthorizations(user, newAuths); } @@ -490,7 +487,7 @@ public class AccumuloInstanceDriver { * Sets the {@link Connector} to this instance. * @param connector the {@link Connector}. */ - public void setConnector(Connector connector) { + public void setConnector(final Connector connector) { this.connector = connector; } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/1d33b435/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloQueryRuleset.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloQueryRuleset.java b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloQueryRuleset.java index f2b5c4c..caae9f5 100644 --- a/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloQueryRuleset.java +++ b/extras/rya.merger/src/main/java/org/apache/rya/accumulo/mr/merge/util/AccumuloQueryRuleset.java @@ -1,24 +1,22 @@ -package org.apache.rya.accumulo.mr.merge.util; - /* - * #%L - * org.apache.rya.accumulo.mr.merge - * %% - * Copyright (C) 2014 Rya - * %% - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * #L% + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. */ +package org.apache.rya.accumulo.mr.merge.util; import java.io.IOException; import java.util.HashMap; @@ -51,9 +49,9 @@ import org.apache.rya.api.utils.NullableStatementImpl; * copying one or more entire tables, independent of the query-derived rules. */ public class AccumuloQueryRuleset extends QueryRuleset { - private Map<TABLE_LAYOUT, List<Range>> tableRanges = new HashMap<>(); - private List<String> entireTables = new LinkedList<String>(); - private RyaTripleContext ryaContext; + private final Map<TABLE_LAYOUT, List<Range>> tableRanges = new HashMap<>(); + private final List<String> entireTables = new LinkedList<String>(); + private final RyaTripleContext ryaContext; /** * Constructs the ruleset and the associated Ranges, given a Configuration that contains a SPARQL query. @@ -61,17 +59,17 @@ public class AccumuloQueryRuleset extends QueryRuleset { * @throws IOException if the range can't be resolved * @throws QueryRulesetException if the query can't be translated to valid rules */ - public AccumuloQueryRuleset(RdfCloudTripleStoreConfiguration conf) throws IOException, QueryRulesetException { + public AccumuloQueryRuleset(final RdfCloudTripleStoreConfiguration conf) throws IOException, QueryRulesetException { // Extract StatementPatterns and conditions from the query super(conf); // Turn StatementPatterns into Ranges ryaContext = RyaTripleContext.getInstance(conf); - for (CopyRule rule : rules) { - StatementPattern sp = rule.getStatement(); - Map.Entry<TABLE_LAYOUT, ByteRange> entry = getRange(sp); - TABLE_LAYOUT layout = entry.getKey(); - ByteRange byteRange = entry.getValue(); - Range range = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); + for (final CopyRule rule : rules) { + final StatementPattern sp = rule.getStatement(); + final Map.Entry<TABLE_LAYOUT, ByteRange> entry = getRange(sp); + final TABLE_LAYOUT layout = entry.getKey(); + final ByteRange byteRange = entry.getValue(); + final Range range = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); if (!tableRanges.containsKey(layout)) { tableRanges.put(layout, new LinkedList<Range>()); } @@ -84,14 +82,14 @@ public class AccumuloQueryRuleset extends QueryRuleset { * @param conf * @throws IOException if the range can't be resolved */ - private Map.Entry<TABLE_LAYOUT, ByteRange> getRange(StatementPattern sp) throws IOException { - Var context = sp.getContextVar(); - Statement stmt = new NullableStatementImpl((Resource) sp.getSubjectVar().getValue(), + private Map.Entry<TABLE_LAYOUT, ByteRange> getRange(final StatementPattern sp) throws IOException { + final Var context = sp.getContextVar(); + final Statement stmt = new NullableStatementImpl((Resource) sp.getSubjectVar().getValue(), (URI) sp.getPredicateVar().getValue(), sp.getObjectVar().getValue(), context == null ? null : (Resource) context.getValue()); - RyaStatement rs = RdfToRyaConversions.convertStatement(stmt); - TriplePatternStrategy strategy = ryaContext.retrieveStrategy(rs); - Map.Entry<TABLE_LAYOUT, ByteRange> entry = + final RyaStatement rs = RdfToRyaConversions.convertStatement(stmt); + final TriplePatternStrategy strategy = ryaContext.retrieveStrategy(rs); + final Map.Entry<TABLE_LAYOUT, ByteRange> entry = strategy.defineRange(rs.getSubject(), rs.getPredicate(), rs.getObject(), rs.getContext(), conf); return entry; } @@ -99,7 +97,7 @@ public class AccumuloQueryRuleset extends QueryRuleset { /** * Add an instruction to select an entire table, with no restricting rule. */ - public void addTable(String tableName) { + public void addTable(final String tableName) { entireTables.add(tableName); } @@ -108,16 +106,16 @@ public class AccumuloQueryRuleset extends QueryRuleset { * @return A Map representing each table and {@link InputTableConfig} needed to get all the rows that match the rules. */ public Map<String, InputTableConfig> getInputConfigs() { - Map<String, InputTableConfig> configs = new HashMap<>(); - for (TABLE_LAYOUT layout : tableRanges.keySet()) { - String parentTable = RdfCloudTripleStoreUtils.layoutPrefixToTable(layout, conf.getTablePrefix()); - InputTableConfig config = new InputTableConfig(); + final Map<String, InputTableConfig> configs = new HashMap<>(); + for (final TABLE_LAYOUT layout : tableRanges.keySet()) { + final String parentTable = RdfCloudTripleStoreUtils.layoutPrefixToTable(layout, conf.getTablePrefix()); + final InputTableConfig config = new InputTableConfig(); config.setRanges(tableRanges.get(layout)); configs.put(parentTable, config); } - for (String tableName : entireTables) { - InputTableConfig config = new InputTableConfig(); - List<Range> ranges = new LinkedList<>(); + for (final String tableName : entireTables) { + final InputTableConfig config = new InputTableConfig(); + final List<Range> ranges = new LinkedList<>(); ranges.add(new Range()); config.setRanges(ranges); configs.put(tableName, config); @@ -134,20 +132,20 @@ public class AccumuloQueryRuleset extends QueryRuleset { * @return Any rules in this ruleset that match the given table and contain the given range * @throws IOException if the Range can't be resolved */ - public List<CopyRule> getRules(TABLE_LAYOUT layout, Range range) throws IOException { - List<CopyRule> matchingRules = new LinkedList<>(); - for (CopyRule rule : rules) { + public List<CopyRule> getRules(final TABLE_LAYOUT layout, final Range range) throws IOException { + final List<CopyRule> matchingRules = new LinkedList<>(); + for (final CopyRule rule : rules) { // Compare the rule to the given range - Map.Entry<TABLE_LAYOUT, ByteRange> entry = getRange(rule.getStatement()); - TABLE_LAYOUT ruleLayout = entry.getKey(); + final Map.Entry<TABLE_LAYOUT, ByteRange> entry = getRange(rule.getStatement()); + final TABLE_LAYOUT ruleLayout = entry.getKey(); // If they apply to different tables, they are unrelated. if (!ruleLayout.equals(layout)) { continue; } // If the given range is contained in (or equal to) the rule's range, then the // rule matches and should be included. - ByteRange byteRange = entry.getValue(); - Range ruleRange = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); + final ByteRange byteRange = entry.getValue(); + final Range ruleRange = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); if (rangeContainsRange(ruleRange, range)) { matchingRules.add(rule); } @@ -157,21 +155,21 @@ public class AccumuloQueryRuleset extends QueryRuleset { @Override public String toString() { - StringBuilder sb = new StringBuilder(super.toString()); - for (String fullTableName : entireTables) { + final StringBuilder sb = new StringBuilder(super.toString()); + for (final String fullTableName : entireTables) { sb.append("\n\tCopy entire table ").append(fullTableName).append("\n"); } return sb.toString(); } - private static boolean rangeContainsRange(Range r1, Range r2) { + private static boolean rangeContainsRange(final Range r1, final Range r2) { // 1. If r1.start is infinite, r1 contains r2.start if (!r1.isInfiniteStartKey()) { // 2. Otherwise, if r2.start is infinite, r1 can't contain r2 if (r2.isInfiniteStartKey()) { return false; } - Key start2 = r2.getStartKey(); + final Key start2 = r2.getStartKey(); // 3. If r2 is inclusive, r1 needs to contain r2's start key. if (r2.isStartKeyInclusive()) { if (!r1.contains(start2)) { @@ -188,7 +186,7 @@ public class AccumuloQueryRuleset extends QueryRuleset { if (r2.isInfiniteStopKey()) { return false; } - Key end2 = r2.getEndKey(); + final Key end2 = r2.getEndKey(); if (r2.isEndKeyInclusive()) { if (!r1.contains(end2)) { return false;
