gjacoby126 commented on a change in pull request #469: PHOENIX-5156 Consistent
Global Indexes for Non-Transactional Tables
URL: https://github.com/apache/phoenix/pull/469#discussion_r274188632
##########
File path:
phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
##########
@@ -579,153 +633,83 @@ public void
postBatchMutateIndispensably(ObserverContext<RegionCoprocessorEnviro
private void doPost(ObserverContext<RegionCoprocessorEnvironment> c,
BatchMutateContext context) throws IOException {
try {
- doPostWithExceptions(c,context);
- return;
- } catch (Throwable e) {
- rethrowIndexingException(e);
- }
- throw new RuntimeException(
- "Somehow didn't complete the index update, but didn't return
succesfully either!");
- }
-
- private void
doPostWithExceptions(ObserverContext<RegionCoprocessorEnvironment> c,
BatchMutateContext context)
- throws IOException {
- //short circuit, if we don't need to do any work
- if (context == null || context.indexUpdates.isEmpty()) {
- return;
- }
-
- // get the current span, or just use a null-span to avoid a bunch of if
statements
- try (TraceScope scope = Trace.startSpan("Completing index writes")) {
- Span current = scope.getSpan();
- if (current == null) {
- current = NullSpan.INSTANCE;
- }
- long start = EnvironmentEdgeManager.currentTimeMillis();
-
- current.addTimelineAnnotation("Actually doing index update for first
time");
- writer.writeAndKillYourselfOnFailure(context.indexUpdates, false,
context.clientVersion);
-
- long duration = EnvironmentEdgeManager.currentTimeMillis() - start;
- if (duration >= slowIndexWriteThreshold) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(getCallTooSlowMessage("indexWrite", duration,
slowIndexWriteThreshold));
- }
- metricSource.incrementNumSlowIndexWriteCalls();
- }
- metricSource.updateIndexWriteTime(duration);
- }
- }
-
- /**
- * Search the {@link WALEdit} for the first {@link IndexedKeyValue} present
- * @param edit {@link WALEdit}
- * @return the first {@link IndexedKeyValue} in the {@link WALEdit} or
<tt>null</tt> if not
- * present
- */
- private IndexedKeyValue getFirstIndexedKeyValue(WALEdit edit) {
- for (Cell kv : edit.getCells()) {
- if (kv instanceof IndexedKeyValue) {
- return (IndexedKeyValue) kv;
- }
- }
- return null;
+ doPostWithExceptions(c,context);
+ return;
+ } catch (Throwable e) {
+ rethrowIndexingException(e);
+ }
+ throw new RuntimeException(
+ "Somehow didn't complete the index update, but didn't return
succesfully either!");
}
- /**
- * Extract the index updates from the WAL Edit
- * @param edit to search for index updates
- * @return the mutations to apply to the index tables
- */
- private Collection<Pair<Mutation, byte[]>> extractIndexUpdate(WALEdit edit) {
- // Avoid multiple internal array resizings. Initial size of 64, unless we
have fewer cells in the edit
- int initialSize = Math.min(edit.size(), 64);
- Collection<Pair<Mutation, byte[]>> indexUpdates = new
ArrayList<Pair<Mutation, byte[]>>(initialSize);
- for (Cell kv : edit.getCells()) {
- if (kv instanceof IndexedKeyValue) {
- IndexedKeyValue ikv = (IndexedKeyValue) kv;
- indexUpdates.add(new Pair<Mutation, byte[]>(ikv.getMutation(),
ikv.getIndexTable()));
- }
- }
+ private void
doPostWithExceptions(ObserverContext<RegionCoprocessorEnvironment> c,
BatchMutateContext context)
+ throws IOException {
+ //short circuit, if we don't need to do any work
+ if (context == null || context.postIndexUpdates.isEmpty()) {
+ return;
+ }
- return indexUpdates;
- }
+ // get the current span, or just use a null-span to avoid a bunch of
if statements
+ try (TraceScope scope = Trace.startSpan("Completing index writes")) {
+ Span current = scope.getSpan();
+ if (current == null) {
+ current = NullSpan.INSTANCE;
+ }
+ long start = EnvironmentEdgeManager.currentTimeMillis();
- @Override
- public void postOpen(final ObserverContext<RegionCoprocessorEnvironment> c) {
- Multimap<HTableInterfaceReference, Mutation> updates =
failedIndexEdits.getEdits(c.getEnvironment().getRegion());
-
- if (this.disabled) {
- return;
- }
+ current.addTimelineAnnotation("Actually doing post index update
for first time");
+ postWriter.writeAndHandleFailure(context.postIndexUpdates, false,
context.clientVersion);
Review comment:
perhaps a comment here to make clear we're writing the verified flags
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
