apache · tkhurana · Dec 17, 2020
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexRepairRegionScannerIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexRepairRegionScannerIT.java
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
@@ -17,11 +17,6 @@
  */
 package org.apache.phoenix.end2end;
 
-import org.apache.phoenix.compile.ExplainPlan;
-import org.apache.phoenix.compile.ExplainPlanAttributes;
-import org.apache.phoenix.jdbc.PhoenixPreparedStatement;
-import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
-import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
@@ -41,18 +36,24 @@
 import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
 import org.apache.hadoop.hbase.regionserver.RegionScanner;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.mapreduce.CounterGroup;
 import org.apache.hadoop.mapreduce.Job;
-import org.apache.phoenix.mapreduce.index.IndexVerificationOutputRepository;
-import org.apache.phoenix.mapreduce.index.IndexVerificationResultRepository;
+import org.apache.phoenix.compile.ExplainPlan;
+import org.apache.phoenix.compile.ExplainPlanAttributes;
 import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixPreparedStatement;
 import org.apache.phoenix.mapreduce.index.IndexTool;
+import org.apache.phoenix.mapreduce.index.IndexVerificationOutputRepository;
+import org.apache.phoenix.mapreduce.index.IndexVerificationResultRepository;
 import org.apache.phoenix.mapreduce.index.PhoenixIndexImportDirectMapper;
+import org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters;
 import org.apache.phoenix.mapreduce.index.PhoenixServerBuildIndexMapper;
 import org.apache.phoenix.query.ConnectionQueryServices;
 import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.query.QueryServicesOptions;
-import org.apache.phoenix.schema.PIndexState;
 import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
 import org.apache.phoenix.transaction.PhoenixTransactionProvider.Feature;
 import org.apache.phoenix.transaction.TransactionFactory;
 import org.apache.phoenix.util.PhoenixRuntime;
@@ -69,6 +70,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
@@ -85,10 +87,10 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.apache.phoenix.mapreduce.PhoenixJobCounters.INPUT_RECORDS;
-import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT_COZ_EXTRA_CELLS;
-import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT_COZ_MISSING_CELLS;
 import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_EXPIRED_INDEX_ROW_COUNT;
 import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT;
+import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT_COZ_EXTRA_CELLS;
+import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_INVALID_INDEX_ROW_COUNT_COZ_MISSING_CELLS;
 import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_MISSING_INDEX_ROW_COUNT;
 import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.BEFORE_REBUILD_VALID_INDEX_ROW_COUNT;
 import static org.apache.phoenix.mapreduce.index.PhoenixIndexToolJobCounters.REBUILT_INDEX_ROW_COUNT;
@@ -781,19 +783,23 @@ public static void assertExplainPlan(boolean localIndex, String actualExplainPla
             actualExplainPlan.contains(expectedExplainPlan));
     }
 
+    public static CounterGroup getMRJobCounters(IndexTool indexTool) throws IOException {
+        return indexTool.getJob().getCounters().getGroup(PhoenixIndexToolJobCounters.class.getName());
+    }
+
     private static List<String> getArgList (boolean directApi, boolean useSnapshot, String schemaName,
                                             String dataTable, String indxTable, String tenantId,
                                             IndexTool.IndexVerifyType verifyType, Long startTime,
                                             Long endTime, Long incrementalVerify) {
         return getArgList(directApi, useSnapshot, schemaName, dataTable, indxTable, tenantId,
-            verifyType, startTime, endTime, IndexTool.IndexDisableLoggingType.NONE, incrementalVerify);
+            verifyType, startTime, endTime, IndexTool.IndexDisableLoggingType.NONE, incrementalVerify, false);
     }
 
     private static List<String> getArgList (boolean directApi, boolean useSnapshot, String schemaName,
                             String dataTable, String indxTable, String tenantId,
                             IndexTool.IndexVerifyType verifyType, Long startTime, Long endTime,
                                             IndexTool.IndexDisableLoggingType disableLoggingType,
-                                            Long incrementalVerify) {
+                                            Long incrementalVerify, boolean useIndexTableAsSource) {
         List<String> args = Lists.newArrayList();
         if (schemaName != null) {
             args.add("--schema=" + schemaName);
@@ -835,6 +841,11 @@ private static List<String> getArgList (boolean directApi, boolean useSnapshot,
             args.add("-rv");
             args.add(String.valueOf(incrementalVerify));
         }
+
+        if (useIndexTableAsSource) {
+            args.add("-fi");
+        }
+
         args.add("-op");
         args.add("/tmp/" + UUID.randomUUID().toString());
         return args;
@@ -851,7 +862,7 @@ public static String[] getArgValues(boolean directApi, boolean useSnapshot, Stri
                                         String dataTable, String indexTable, String tenantId, IndexTool.IndexVerifyType verifyType,
                                         IndexTool.IndexDisableLoggingType disableLoggingType) {
         List<String> args = getArgList(directApi, useSnapshot, schemaName, dataTable, indexTable,
-                tenantId, verifyType, null, null, disableLoggingType, null);
+                tenantId, verifyType, null, null, disableLoggingType, null, false);
         return args.toArray(new String[0]);
     }
 
@@ -870,7 +881,19 @@ public static String[] getArgValues(boolean directApi, boolean useSnapshot, Stri
                                          IndexTool.IndexDisableLoggingType disableLoggingType,
                                          Long incrementalVerify ) {
         List<String> args = getArgList(directApi, useSnapshot, schemaName, dataTable, indexTable,
-            tenantId, verifyType, startTime, endTime, disableLoggingType, incrementalVerify);
+            tenantId, verifyType, startTime, endTime, disableLoggingType, incrementalVerify, false);
+        return args.toArray(new String[0]);
+    }
+
+    public static String [] getArgValues(boolean directApi, boolean useSnapshot, String schemaName,
+        String dataTable, String indexTable, String tenantId,
+        IndexTool.IndexVerifyType verifyType, Long startTime,
+        Long endTime,
+        IndexTool.IndexDisableLoggingType disableLoggingType,
+        Long incrementalVerify,
+        boolean useIndexTableAsSource) {
+        List<String> args = getArgList(directApi, useSnapshot, schemaName, dataTable, indexTable,
+            tenantId, verifyType, startTime, endTime, disableLoggingType, incrementalVerify, useIndexTableAsSource);
         return args.toArray(new String[0]);
     }
 

diff --git a/...x-core/src/it/java/org/apache/phoenix/mapreduce/PhoenixServerBuildIndexInputFormatIT.java b/...x-core/src/it/java/org/apache/phoenix/mapreduce/PhoenixServerBuildIndexInputFormatIT.java
@@ -0,0 +1,105 @@
+/*
+ * 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.mapreduce;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.phoenix.compile.QueryPlan;
+import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
+import org.apache.phoenix.mapreduce.index.IndexScrutinyTool.SourceTable;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.SchemaUtil;
+import org.junit.Test;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.Properties;
+
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+import static org.junit.Assert.assertEquals;
+
+public class PhoenixServerBuildIndexInputFormatIT  extends ParallelStatsDisabledIT {
+
+    @Test
+    public void testQueryPlanWithSource() throws Exception {
+        PhoenixServerBuildIndexInputFormat inputFormat;
+        Configuration conf = new Configuration(getUtility().getConfiguration());
+        String schemaName = generateUniqueName();
+        String dataTableName = generateUniqueName();
+        String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName);
+        String indexTableName = generateUniqueName();
+        String indexTableFullName = SchemaUtil.getTableName(schemaName, indexTableName);
+        String viewName = generateUniqueName();
+        String viewFullName = SchemaUtil.getTableName(schemaName, viewName);
+        String viewIndexName = generateUniqueName();
+        String viewIndexFullName = SchemaUtil.getTableName(schemaName, viewIndexName);
+        Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+        try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+            conn.createStatement().execute("CREATE TABLE " + dataTableFullName
+                + " (ID INTEGER NOT NULL PRIMARY KEY, VAL1 INTEGER, VAL2 INTEGER) ");
+            conn.createStatement().execute(String.format(
+                "CREATE INDEX %s ON %s (VAL1) INCLUDE (VAL2)", indexTableName, dataTableFullName));
+            conn.createStatement().execute("CREATE VIEW " + viewFullName +
+                " AS SELECT * FROM " + dataTableFullName);
+            conn.commit();
+
+            conn.createStatement().execute(String.format(
+                "CREATE INDEX %s ON %s (VAL1) INCLUDE (VAL2)", viewIndexName, viewFullName));
+
+            PhoenixConfigurationUtil.setIndexToolDataTableName(conf, dataTableFullName);
+            PhoenixConfigurationUtil.setIndexToolIndexTableName(conf, indexTableFullName);
+            // use data table as source (default)
+            assertTableSource(conf, conn);
+
+            // use index table as source
+            PhoenixConfigurationUtil.setIndexToolSourceTable(conf, SourceTable.INDEX_TABLE_SOURCE);
+            assertTableSource(conf, conn);
+
+            PhoenixConfigurationUtil.setIndexToolDataTableName(conf, viewFullName);
+            PhoenixConfigurationUtil.setIndexToolIndexTableName(conf, viewIndexFullName);
+            PhoenixConfigurationUtil.setIndexToolSourceTable(conf, SourceTable.DATA_TABLE_SOURCE);
+
+            assertTableSource(conf, conn);
+
+            PhoenixConfigurationUtil.setIndexToolSourceTable(conf, SourceTable.INDEX_TABLE_SOURCE);
+            assertTableSource(conf, conn);
+        }
+    }
+
+    private void assertTableSource(Configuration conf, Connection conn) throws Exception {
+        String dataTableFullName = PhoenixConfigurationUtil.getIndexToolDataTableName(conf);
+        String indexTableFullName = PhoenixConfigurationUtil.getIndexToolIndexTableName(conf);
+        SourceTable sourceTable = PhoenixConfigurationUtil.getIndexToolSourceTable(conf);
+        boolean fromIndex = sourceTable.equals(SourceTable.INDEX_TABLE_SOURCE);
+        PTable pDataTable = PhoenixRuntime.getTable(conn, dataTableFullName);
+        PTable pIndexTable = PhoenixRuntime.getTable(conn, indexTableFullName);
+
+        PhoenixServerBuildIndexInputFormat inputFormat = new PhoenixServerBuildIndexInputFormat();
+        QueryPlan queryPlan = inputFormat.getQueryPlan(Job.getInstance(), conf);
+        PTable actual = queryPlan.getTableRef().getTable();
+
+        if (!fromIndex) {
+            assertEquals(pDataTable, actual);
+        } else {
+            assertEquals(pIndexTable, actual);
+        }
+    }
+}