Updated Branches: refs/heads/master acadf0611 -> fd208badc
Fixed BLUR-98 Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/77262a9d Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/77262a9d Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/77262a9d Branch: refs/heads/master Commit: 77262a9d135dea1c03ef184e2ca20e502baf0814 Parents: acadf06 Author: Aaron McCurry <[email protected]> Authored: Mon Jun 17 15:36:47 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Mon Jun 17 15:36:47 2013 -0400 ---------------------------------------------------------------------- .../blur/manager/writer/IndexImporter.java | 2 + .../apache/blur/mapreduce/lib/BlurCounters.java | 2 +- .../apache/blur/mapreduce/lib/BlurMutate.java | 33 +++++++++++--- .../blur/mapreduce/lib/BlurOutputFormat.java | 48 +++++++++++++++----- .../org/apache/blur/utils/BlurConstants.java | 2 + 5 files changed, 69 insertions(+), 18 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/77262a9d/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java ---------------------------------------------------------------------- diff --git a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java index e4664af..5f7d148 100644 --- a/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java +++ b/blur-core/src/main/java/org/apache/blur/manager/writer/IndexImporter.java @@ -125,6 +125,8 @@ public class IndexImporter extends TimerTask implements Closeable { if (isSuccess) { LOG.info("Add index [{0}] [{1}/{2}]", directory, shard, table); indexWriter.addIndexes(directory); + LOG.info("Removing delete markers [{0}] on [{1}/{2}]", directory, shard, table); + indexWriter.deleteDocuments(new Term(BlurConstants.DELETE_MARKER, BlurConstants.DELETE_MARKER_VALUE)); LOG.info("Finishing import [{0}], commiting on [{1}/{2}]", directory, shard, table); indexWriter.commit(); LOG.info("Cleaning up old directory [{0}] for [{1}/{2}]", dirPath, shard, table); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/77262a9d/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurCounters.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurCounters.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurCounters.java index ba8da67..4d008e7 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurCounters.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurCounters.java @@ -21,6 +21,6 @@ package org.apache.blur.mapreduce.lib; * The enum class used for all the internal counters during map reduce jobs. */ public enum BlurCounters { - RECORD_COUNT, FIELD_COUNT, ROW_COUNT, RECORD_RATE, COPY_RATE, ROW_RATE, RECORD_DUPLICATE_COUNT, ROW_OVERFLOW_COUNT + RECORD_COUNT, FIELD_COUNT, ROW_COUNT, RECORD_RATE, COPY_RATE, ROW_RATE, RECORD_DUPLICATE_COUNT, ROW_OVERFLOW_COUNT, ROW_DELETE_COUNT } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/77262a9d/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMutate.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMutate.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMutate.java index 3fdb97c..36d7f4f 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMutate.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurMutate.java @@ -20,12 +20,28 @@ import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; +import org.apache.blur.thrift.generated.Record; +import org.apache.blur.thrift.generated.Row; import org.apache.hadoop.io.Writable; +/** + * {@link BlurMutate} carries the {@link Record}s bound for the {@link Row} for + * indexing. If this mutate represents a delete of the {@link Row} the recordId + * of the {@link BlurRecord} is ignored. + */ public class BlurMutate implements Writable { + /** + * The {@link MUTATE_TYPE} controls the mutating of the {@link Row}. DELETE + * indicates that the {@link Row} is to be deleted. REPLACE indicates that the + * group of mutates are to replace the existing {@link Row}. + * + * If both a DELETE and a REPLACE exist for a single {@link Row} in the + * {@link BlurOutputFormat} then the {@link Row} will be replaced not just + * deleted. + */ public enum MUTATE_TYPE { - /*ADD(0), UPDATE(1),*/ DELETE(2), REPLACE(3); + /* ADD(0), UPDATE(1), */DELETE(2), REPLACE(3); private int _value; private MUTATE_TYPE(int value) { @@ -38,11 +54,11 @@ public class BlurMutate implements Writable { public MUTATE_TYPE find(int value) { switch (value) { - // @TODO Updates through MR is going to be disabled -// case 0: -// return ADD; -// case 1: -// return UPDATE; + // @TODO Updates through MR is going to be disabled + // case 0: + // return ADD; + // case 1: + // return UPDATE; case 2: return DELETE; case 3: @@ -65,6 +81,11 @@ public class BlurMutate implements Writable { _record = record; } + public BlurMutate(MUTATE_TYPE type, String rowId) { + _mutateType = type; + _record.setRowId(rowId); + } + public BlurMutate(MUTATE_TYPE type, String rowId, String recordId) { _mutateType = type; _record.setRowId(rowId); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/77262a9d/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java ---------------------------------------------------------------------- diff --git a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java index ae91eb8..174c477 100644 --- a/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java +++ b/blur-mapred/src/main/java/org/apache/blur/mapreduce/lib/BlurOutputFormat.java @@ -29,6 +29,7 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.lucene.LuceneVersionConstant; import org.apache.blur.manager.writer.TransactionRecorder; +import org.apache.blur.mapreduce.lib.BlurMutate.MUTATE_TYPE; import org.apache.blur.store.hdfs.HdfsDirectory; import org.apache.blur.thirdparty.thrift_0_9_0.TException; import org.apache.blur.thirdparty.thrift_0_9_0.protocol.TJSONProtocol; @@ -123,8 +124,7 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { } @Override - public void checkOutputSpecs(JobContext context) throws IOException, - InterruptedException { + public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException { Configuration config = context.getConfiguration(); TableDescriptor tableDescriptor = getTableDescriptor(config); if (tableDescriptor == null) { @@ -133,17 +133,19 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { int shardCount = tableDescriptor.getShardCount(); FileSystem fileSystem = getOutputPath(config).getFileSystem(config); Path tablePath = new Path(tableDescriptor.getTableUri()); - if(fileSystem.exists(tablePath)) { + if (fileSystem.exists(tablePath)) { BlurUtil.validateShardCount(shardCount, fileSystem, tablePath); - }else{ - throw new IOException("Table path [ "+ tablePath + " ] doesn't exist for table [ " + tableDescriptor.getName() + " ]."); + } else { + throw new IOException("Table path [ " + tablePath + " ] doesn't exist for table [ " + tableDescriptor.getName() + + " ]."); } int reducers = context.getNumReduceTasks(); int reducerMultiplier = getReducerMultiplier(config); int validNumberOfReducers = reducerMultiplier * shardCount; if (reducers > 0 && reducers != validNumberOfReducers) { - throw new IllegalArgumentException("Invalid number of reducers [ " + reducers +" ]." + " Number of Reducers should be [ " + validNumberOfReducers + " ]."); + throw new IllegalArgumentException("Invalid number of reducers [ " + reducers + " ]." + + " Number of Reducers should be [ " + validNumberOfReducers + " ]."); } } @@ -199,7 +201,7 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { Configuration configuration = job.getConfiguration(); configuration.setInt(BLUR_OUTPUT_REDUCER_MULTIPLIER, multiple); } - + public static int getReducerMultiplier(Configuration configuration) { return configuration.getInt(BLUR_OUTPUT_REDUCER_MULTIPLIER, 1); } @@ -364,6 +366,9 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { private File _localTmpPath; private ProgressableDirectory _localTmpDir; private Counter _rowOverFlowCount; + private String _deletedRowId; + + private Counter _rowDeleteCount; public BlurRecordWriter(Configuration configuration, BlurAnalyzer blurAnalyzer, int attemptId, String tmpDirName) throws IOException { @@ -420,6 +425,7 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { _recordCount = getCounter.getCounter(BlurCounters.RECORD_COUNT); _recordDuplicateCount = getCounter.getCounter(BlurCounters.RECORD_DUPLICATE_COUNT); _rowCount = getCounter.getCounter(BlurCounters.ROW_COUNT); + _rowDeleteCount = getCounter.getCounter(BlurCounters.ROW_DELETE_COUNT); _rowOverFlowCount = getCounter.getCounter(BlurCounters.ROW_OVERFLOW_COUNT); _recordRateCounter = new RateCounter(getCounter.getCounter(BlurCounters.RECORD_RATE)); _rowRateCounter = new RateCounter(getCounter.getCounter(BlurCounters.ROW_RATE)); @@ -430,6 +436,10 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { BlurRecord blurRecord = value.getRecord(); Record record = getRecord(blurRecord); String recordId = record.getRecordId(); + if (value.getMutateType() == MUTATE_TYPE.DELETE) { + _deletedRowId = blurRecord.getRowId(); + return; + } Document document = TransactionRecorder.convert(blurRecord.getRowId(), record, _builder, _analyzer); if (_documents.size() == 0) { document.add(new StringField(BlurConstants.PRIME_DOC, BlurConstants.PRIME_DOC_VALUE, Store.NO)); @@ -485,6 +495,8 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { private void flush() throws CorruptIndexException, IOException { if (_usingLocalTmpindex) { + // since we have flushed to disk then we do not need to index the + // delete. flushToTmpIndex(); _localTmpWriter.close(false); DirectoryReader reader = DirectoryReader.open(_localTmpDir); @@ -495,16 +507,30 @@ public class BlurOutputFormat extends OutputFormat<Text, BlurMutate> { _rowOverFlowCount.increment(1); } else { if (_documents.isEmpty()) { - return; + if (_deletedRowId != null) { + _writer.addDocument(getDeleteDoc()); + _rowDeleteCount.increment(1); + } else { + LOG.info("This case should never happen, no records to index and no row deletes to emit."); + } + } else { + _writer.addDocuments(_documents.values()); + _recordRateCounter.mark(_documents.size()); + _documents.clear(); } - _writer.addDocuments(_documents.values()); - _recordRateCounter.mark(_documents.size()); - _documents.clear(); } + _deletedRowId = null; _rowRateCounter.mark(); _rowCount.increment(1); } + private Document getDeleteDoc() { + Document document = new Document(); + document.add(new StringField(BlurConstants.ROW_ID, _deletedRowId, Store.NO)); + document.add(new StringField(BlurConstants.DELETE_MARKER, BlurConstants.DELETE_MARKER_VALUE, Store.NO)); + return document; + } + @Override public void close(TaskAttemptContext context) throws IOException, InterruptedException { flush(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/77262a9d/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java ---------------------------------------------------------------------- diff --git a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java index e929f83..e797073 100644 --- a/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java +++ b/blur-util/src/main/java/org/apache/blur/utils/BlurConstants.java @@ -89,6 +89,8 @@ public class BlurConstants { public static final String BLUR_CLUSTER; public static final long ZK_WAIT_TIME = TimeUnit.SECONDS.toMillis(5); + public static final String DELETE_MARKER_VALUE = "delete"; + public static final String DELETE_MARKER = "_deletemarker_"; static { try {
