PHOENIX-7709 Index committer post writer lazy mode - Async RPC call for verified index mutations (#2297)

Author: virajjasani

phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/parallel/BaseTaskRunner.java

@@ -116,6 +116,13 @@ public <R> Pair<List<R>, List<Future<R>>> submitUninterruptible(TaskBatch<R> tas
     throw new EarlyExitFailure("Interrupted and stopped before computation was complete!");
   }
 
+  @Override
+  public <R> void submitOnly(TaskBatch<R> tasks) {
+    for (Task<R> task : tasks.getTasks()) {
+      this.writerPool.submit(task);
+    }
+  }
+
   @Override
   public void stop(String why) {
     if (this.stopped) {

phoenix-core-client/src/main/java/org/apache/phoenix/hbase/index/parallel/TaskRunner.java

@@ -60,4 +60,23 @@ public <R> Pair<List<R>, List<Future<R>>> submit(TaskBatch<R> tasks)
    */
   public <R> Pair<List<R>, List<Future<R>>> submitUninterruptible(TaskBatch<R> tasks)
     throws EarlyExitFailure, ExecutionException;
+
+  /**
+   * Submit the given tasks to the pool without waiting for them to complete or collecting results.
+   * This is a fire-and-forget operation that allows tasks to run asynchronously in the background.
+   * <p>
+   * Unlike {@link #submit(TaskBatch)} and {@link #submitUninterruptible(TaskBatch)}, this method
+   * does not block waiting for task completion and does not return results or futures. It is useful
+   * for scenarios where you want to initiate background processing but don't need to wait for or
+   * collect the results.
+   * <p>
+   * Tasks are submitted to the underlying thread pool and will execute according to the pool's
+   * scheduling policy. If any task fails during execution, the failure will be handled internally
+   * and will not propagate back to the caller since no results are collected.
+   * @param <R>   the type of result that would be returned by the tasks (unused since no results
+   *              are collected)
+   * @param tasks the batch of tasks to submit for asynchronous execution
+   * @throws ExecutionException if there is an error submitting the tasks to the thread pool
+   */
+  <R> void submitOnly(TaskBatch<R> tasks) throws ExecutionException;
 }

phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/AbstractParallelWriterIndexCommitter.java

@@ -63,7 +63,6 @@ public abstract class AbstractParallelWriterIndexCommitter implements IndexCommi
   protected QuickFailingTaskRunner pool;
   protected KeyValueBuilder kvBuilder;
   protected RegionCoprocessorEnvironment env;
-  protected TaskBatch<Void> tasks;
   protected boolean disableIndexOnFailure = false;
 
   // This relies on Hadoop Configuration to handle warning about deprecated configs and
@@ -116,6 +115,39 @@ public void setup(HTableFactory factory, ExecutorService pool, Stoppable stop,
   public void write(Multimap<HTableInterfaceReference, Mutation> toWrite,
     final boolean allowLocalUpdates, final int clientVersion)
     throws SingleIndexWriteFailureException {
+    TaskBatch<Void> tasks = new TaskBatch<>(toWrite.asMap().size());
+    addTasks(toWrite, allowLocalUpdates, clientVersion, tasks);
+    submitTasks(tasks);
+  }
+
+  /**
+   * Submits the provided task batch for execution. This method defines the task submission strategy
+   * and must be implemented by concrete subclasses to specify whether tasks should be executed
+   * synchronously (blocking until completion) or asynchronously (fire-and-forget).
+   * @param tasks the batch of index write tasks to submit for execution
+   * @throws SingleIndexWriteFailureException if there is an error during task submission or
+   *                                          execution (implementation-dependent)
+   */
+  protected abstract void submitTasks(TaskBatch<Void> tasks)
+    throws SingleIndexWriteFailureException;
+
+  /**
+   * Adds parallel index write tasks to the provided task batch for execution across multiple index
+   * tables. Each index table gets its own task that will be executed in parallel to optimize write
+   * performance.
+   * @param toWrite           a multimap containing index table references as keys and their
+   *                          corresponding mutations as values. Each table will get its own
+   *                          parallel task.
+   * @param allowLocalUpdates if false, skips creating tasks for writes to the same table as the
+   *                          current region to prevent potential deadlocks
+   * @param clientVersion     the Phoenix client version, used for compatibility checks and
+   *                          version-specific behavior in the index write operations
+   * @param tasks             the task batch to which the newly created index write tasks will be
+   *                          added. This batch needs to be submitted for parallel execution by the
+   *                          caller.
+   */
+  private void addTasks(Multimap<HTableInterfaceReference, Mutation> toWrite,
+    boolean allowLocalUpdates, int clientVersion, TaskBatch<Void> tasks) {
     /*
      * This bit here is a little odd, so let's explain what's going on. Basically, we want to do the
      * writes in parallel to each index table, so each table gets its own task and is submitted to
@@ -128,7 +160,6 @@ public void write(Multimap<HTableInterfaceReference, Mutation> toWrite,
      */
 
     Set<Entry<HTableInterfaceReference, Collection<Mutation>>> entries = toWrite.asMap().entrySet();
-    tasks = new TaskBatch<Void>(entries.size());
     for (Entry<HTableInterfaceReference, Collection<Mutation>> entry : entries) {
       // get the mutations for each table. We leak the implementation here a little bit to save
       // doing a complete copy over of all the index update for each table.

phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/LazyParallelWriterIndexCommitter.java

@@ -17,17 +17,31 @@
  */
 package org.apache.phoenix.hbase.index.write;
 
+import org.apache.phoenix.hbase.index.exception.SingleIndexWriteFailureException;
+import org.apache.phoenix.hbase.index.parallel.TaskBatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * Like the {@link ParallelWriterIndexCommitter}, but does not block
  */
 public class LazyParallelWriterIndexCommitter extends AbstractParallelWriterIndexCommitter {
 
-  // for testing
-  public LazyParallelWriterIndexCommitter(String hbaseVersion) {
-    super(hbaseVersion);
-  }
+  private static final Logger LOGGER =
+    LoggerFactory.getLogger(LazyParallelWriterIndexCommitter.class);
 
   public LazyParallelWriterIndexCommitter() {
     super();
   }
+
+  @Override
+  protected void submitTasks(TaskBatch<Void> tasks) throws SingleIndexWriteFailureException {
+    try {
+      pool.submitOnly(tasks);
+    } catch (Exception e) {
+      LOGGER.error("Error while submitting the task.", e);
+      propagateFailure(e.getCause());
+    }
+  }
+
 }

phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/ParallelWriterIndexCommitter.java

@@ -18,15 +18,12 @@
 package org.apache.phoenix.hbase.index.write;
 
 import java.util.concurrent.ExecutionException;
-import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.phoenix.hbase.index.exception.SingleIndexWriteFailureException;
 import org.apache.phoenix.hbase.index.parallel.EarlyExitFailure;
-import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
+import org.apache.phoenix.hbase.index.parallel.TaskBatch;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.phoenix.thirdparty.com.google.common.collect.Multimap;
-
 /**
  * Write index updates to the index tables in parallel. We attempt to early exit from the writes if
  * any of the index updates fails. Completion is determined by the following criteria: *
@@ -49,12 +46,7 @@ public ParallelWriterIndexCommitter(String hbaseVersion) {
   }
 
   @Override
-  public void write(Multimap<HTableInterfaceReference, Mutation> toWrite,
-    final boolean allowLocalUpdates, final int clientVersion)
-    throws SingleIndexWriteFailureException {
-
-    super.write(toWrite, allowLocalUpdates, clientVersion);
-    // actually submit the tasks to the pool and wait for them to finish/fail
+  protected void submitTasks(TaskBatch<Void> tasks) throws SingleIndexWriteFailureException {
     try {
       pool.submitUninterruptible(tasks);
     } catch (EarlyExitFailure e) {
@@ -63,6 +55,5 @@ public void write(Multimap<HTableInterfaceReference, Mutation> toWrite,
       LOGGER.error("Found a failed index update!");
       propagateFailure(e.getCause());
     }
-
   }
 }

phoenix-core/src/it/java/org/apache/phoenix/end2end/Bson4IT.java

@@ -44,6 +44,9 @@
 import org.apache.phoenix.compile.ExplainPlan;
 import org.apache.phoenix.compile.ExplainPlanAttributes;
 import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.hbase.index.metrics.GlobalIndexCheckerSource;
+import org.apache.phoenix.hbase.index.metrics.GlobalIndexCheckerSourceImpl;
+import org.apache.phoenix.hbase.index.metrics.MetricsIndexerSourceFactory;
 import org.apache.phoenix.jdbc.PhoenixPreparedStatement;
 import org.apache.phoenix.jdbc.PhoenixStatement;
 import org.apache.phoenix.schema.types.PDouble;
@@ -68,7 +71,7 @@
 /**
  * Tests for BSON.
  */
-@Category(ParallelStatsDisabledTest.class)
+@Category(NeedsOwnMiniClusterTest.class)
 @RunWith(Parameterized.class)
 public class Bson4IT extends ParallelStatsDisabledIT {
 
@@ -291,7 +294,8 @@ public void testBsonValueWithBinaryEncoded() throws Exception {
     try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
 
       String ddl = "CREATE TABLE " + tableName + " (PK1 VARCHAR NOT NULL, C1 VARCHAR, COL BSON"
-        + " CONSTRAINT pk PRIMARY KEY(PK1))";
+        + " CONSTRAINT pk PRIMARY KEY(PK1))"
+        + " \"org.apache.hadoop.hbase.index.lazy.post_batch.write\"=true";
 
       final String indexDdl1;
       if (!this.coveredIndex) {
@@ -367,6 +371,7 @@ public void testBsonValueWithBinaryEncoded() throws Exception {
       stmt.executeUpdate();
 
       conn.commit();
+      Thread.sleep(500);
 
       ResultSet rs = conn.createStatement().executeQuery("SELECT count(*) FROM " + tableName);
       assertTrue(rs.next());
@@ -450,9 +455,25 @@ public void testBsonValueWithBinaryEncoded() throws Exception {
       assertFalse(rs.next());
 
       validateExplainPlan(ps, tableName, "FULL SCAN ");
+      verifyNoReadRepair();
     }
   }
 
+  private void verifyNoReadRepair() {
+    GlobalIndexCheckerSourceImpl indexCheckerSource =
+      (GlobalIndexCheckerSourceImpl) MetricsIndexerSourceFactory.getInstance()
+        .getGlobalIndexCheckerSource();
+
+    long indexRepairs = indexCheckerSource.getMetricsRegistry()
+      .getCounter(GlobalIndexCheckerSource.INDEX_REPAIR, 0).value();
+    assertEquals("No index repairs should occur during test execution", 0, indexRepairs);
+
+    long indexRepairFailures = indexCheckerSource.getMetricsRegistry()
+      .getCounter(GlobalIndexCheckerSource.INDEX_REPAIR_FAILURE, 0).value();
+    assertEquals("No index repair failures should occur during test execution", 0,
+      indexRepairFailures);
+  }
+
   @Test
   public void testBsonValueFunctionWithBSONType() throws Exception {
     Assume.assumeTrue(this.coveredIndex && this.columnEncoded); // Since indexing on BSON not

phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsExtendedIT.java

@@ -68,7 +68,7 @@ public class ConcurrentMutationsExtendedIT extends ParallelStatsDisabledIT {
   private static final String MVCC_LOCK_TEST_TABLE_PREFIX = "MVCCLOCKTEST_";
   private static final String LOCK_TEST_TABLE_PREFIX = "LOCKTEST_";
   private static final int ROW_LOCK_WAIT_TIME = 10000;
-  private static final int MAX_LOOKBACK_AGE = 1000000;
+  protected static final int MAX_LOOKBACK_AGE = 1000000;
   private final Object lock = new Object();
 
   public ConcurrentMutationsExtendedIT(boolean uncovered) {

phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsLazyPostBatchWriteIT.java

@@ -0,0 +1,53 @@
+/*
+ * 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
+ *
+ * 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.phoenix.end2end;
+
+import java.util.Map;
+import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
+import org.apache.phoenix.hbase.index.IndexRegionObserver;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+
+/**
+ * Test class that extends ConcurrentMutationsExtendedIT with lazy post batch write enabled.
+ */
+@Category(NeedsOwnMiniClusterTest.class)
+public class ConcurrentMutationsLazyPostBatchWriteIT extends ConcurrentMutationsExtendedIT {
+
+  public ConcurrentMutationsLazyPostBatchWriteIT(boolean uncovered) {
+    super(uncovered);
+    Assume.assumeFalse("Only covered index supports lazy post batch write mode", uncovered);
+  }
+
+  @BeforeClass
+  public static synchronized void doSetup() throws Exception {
+    Map<String, String> props = Maps.newHashMapWithExpectedSize(4);
+    props.put(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB, Long.toString(0));
+    props.put(IndexRegionObserver.INDEX_LAZY_POST_BATCH_WRITE, "true");
+    props.put(BaseScannerRegionObserverConstants.PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY,
+      Integer.toString(MAX_LOOKBACK_AGE));
+    props.put("hbase.rowlock.wait.duration", "100");
+    props.put("phoenix.index.concurrent.wait.duration.ms", "10");
+    setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+  }
+}