| /* |
| * 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.coprocessor; |
| |
| import static org.apache.phoenix.schema.types.PDataType.TRUE_BYTES; |
| |
| import java.io.IOException; |
| import java.security.PrivilegedExceptionAction; |
| import java.sql.SQLException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.TimerTask; |
| import java.util.concurrent.ScheduledThreadPoolExecutor; |
| import java.util.concurrent.TimeUnit; |
| |
| import javax.annotation.concurrent.GuardedBy; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.hbase.Cell; |
| import org.apache.hadoop.hbase.CellUtil; |
| import org.apache.hadoop.hbase.CoprocessorEnvironment; |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.client.HTableInterface; |
| import org.apache.hadoop.hbase.client.Result; |
| import org.apache.hadoop.hbase.client.Scan; |
| import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver; |
| import org.apache.hadoop.hbase.coprocessor.ObserverContext; |
| import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; |
| import org.apache.hadoop.hbase.filter.CompareFilter; |
| import org.apache.hadoop.hbase.filter.SingleColumnValueFilter; |
| import org.apache.hadoop.hbase.io.ImmutableBytesWritable; |
| import org.apache.hadoop.hbase.regionserver.RegionScanner; |
| import org.apache.hadoop.hbase.security.User; |
| import org.apache.hadoop.hbase.util.Pair; |
| |
| import org.apache.phoenix.cache.GlobalCache; |
| import org.apache.phoenix.compile.MutationPlan; |
| import org.apache.phoenix.compile.PostDDLCompiler; |
| import org.apache.phoenix.execute.MutationState; |
| import org.apache.phoenix.hbase.index.util.IndexManagementUtil; |
| import org.apache.phoenix.index.IndexMaintainer; |
| import org.apache.phoenix.index.PhoenixIndexCodec; |
| import org.apache.phoenix.jdbc.PhoenixConnection; |
| import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; |
| import org.apache.phoenix.jdbc.PhoenixDriver; |
| import org.apache.phoenix.query.QueryConstants; |
| import org.apache.phoenix.query.QueryServices; |
| import org.apache.phoenix.query.QueryServicesOptions; |
| import org.apache.phoenix.schema.PIndexState; |
| import org.apache.phoenix.schema.PName; |
| import org.apache.phoenix.schema.PTable; |
| import org.apache.phoenix.schema.SortOrder; |
| import org.apache.phoenix.schema.TableRef; |
| import org.apache.phoenix.schema.types.PLong; |
| import org.apache.phoenix.util.ByteUtil; |
| import org.apache.phoenix.util.EnvironmentEdgeManager; |
| import org.apache.phoenix.util.IndexUtil; |
| import org.apache.phoenix.util.MetaDataUtil; |
| import org.apache.phoenix.util.PhoenixRuntime; |
| import org.apache.phoenix.util.PropertiesUtil; |
| import org.apache.phoenix.util.QueryUtil; |
| import org.apache.phoenix.util.ReadOnlyProps; |
| import org.apache.phoenix.util.ScanUtil; |
| import org.apache.phoenix.util.SchemaUtil; |
| import org.apache.phoenix.util.ServerUtil; |
| import org.apache.phoenix.util.UpgradeUtil; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| |
| |
| /** |
| * Coprocessor for metadata related operations. This coprocessor would only be registered |
| * to SYSTEM.TABLE. |
| */ |
| @SuppressWarnings("deprecation") |
| public class MetaDataRegionObserver extends BaseRegionObserver { |
| public static final Logger LOGGER = LoggerFactory.getLogger(MetaDataRegionObserver.class); |
| public static final String REBUILD_INDEX_APPEND_TO_URL_STRING = "REBUILDINDEX"; |
| // PHOENIX-5094 To differentiate the increment in PENDING_DISABLE_COUNT made by client or index |
| // rebuilder, we are using large value for index rebuilder |
| public static final long PENDING_DISABLE_INACTIVE_STATE_COUNT = 10000L; |
| private static final byte[] SYSTEM_CATALOG_KEY = SchemaUtil.getTableKey( |
| ByteUtil.EMPTY_BYTE_ARRAY, |
| QueryConstants.SYSTEM_SCHEMA_NAME_BYTES, |
| PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE_BYTES); |
| protected ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1); |
| private boolean enableRebuildIndex = QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD; |
| private long rebuildIndexTimeInterval = QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_INTERVAL; |
| private static Map<PName, Long> batchExecutedPerTableMap = new HashMap<PName, Long>(); |
| @GuardedBy("MetaDataRegionObserver.class") |
| private static Properties rebuildIndexConnectionProps; |
| // Added for test purposes |
| private long initialRebuildTaskDelay; |
| |
| @Override |
| public void preClose(final ObserverContext<RegionCoprocessorEnvironment> c, |
| boolean abortRequested) { |
| executor.shutdownNow(); |
| GlobalCache.getInstance(c.getEnvironment()).getMetaDataCache().invalidateAll(); |
| } |
| |
| @Override |
| public void start(CoprocessorEnvironment env) throws IOException { |
| // sleep a little bit to compensate time clock skew when SYSTEM.CATALOG moves |
| // among region servers because we relies on server time of RS which is hosting |
| // SYSTEM.CATALOG |
| Configuration config = env.getConfiguration(); |
| long sleepTime = config.getLong(QueryServices.CLOCK_SKEW_INTERVAL_ATTRIB, |
| QueryServicesOptions.DEFAULT_CLOCK_SKEW_INTERVAL); |
| try { |
| if(sleepTime > 0) { |
| Thread.sleep(sleepTime); |
| } |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| } |
| enableRebuildIndex = |
| config.getBoolean( |
| QueryServices.INDEX_FAILURE_HANDLING_REBUILD_ATTRIB, |
| QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD); |
| rebuildIndexTimeInterval = |
| config.getLong( |
| QueryServices.INDEX_FAILURE_HANDLING_REBUILD_INTERVAL_ATTRIB, |
| QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_INTERVAL); |
| initialRebuildTaskDelay = |
| config.getLong( |
| QueryServices.INDEX_REBUILD_TASK_INITIAL_DELAY, |
| QueryServicesOptions.DEFAULT_INDEX_REBUILD_TASK_INITIAL_DELAY); |
| } |
| |
| @Override |
| public void postOpen(ObserverContext<RegionCoprocessorEnvironment> e) { |
| final RegionCoprocessorEnvironment env = e.getEnvironment(); |
| |
| Runnable r = new Runnable() { |
| @Override |
| public void run() { |
| HTableInterface metaTable = null; |
| HTableInterface statsTable = null; |
| try { |
| ReadOnlyProps props=new ReadOnlyProps(env.getConfiguration().iterator()); |
| Thread.sleep(1000); |
| metaTable = env.getTable( |
| SchemaUtil.getPhysicalName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, props)); |
| statsTable = env.getTable( |
| SchemaUtil.getPhysicalName(PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES, props)); |
| final HTableInterface mTable=metaTable; |
| final HTableInterface sTable=statsTable; |
| User.runAsLoginUser(new PrivilegedExceptionAction<Void>() { |
| @Override |
| public Void run() throws Exception { |
| if (UpgradeUtil.truncateStats(mTable, sTable)) { |
| LOGGER.info("Stats are successfully truncated for upgrade 4.7!!"); |
| } |
| return null; |
| } |
| }); |
| |
| } catch (Exception exception) { |
| LOGGER.warn("Exception while truncate stats.., please check and delete stats " + |
| "manually inorder to get proper result with old client!!"); |
| LOGGER.warn(exception.getStackTrace().toString()); |
| } finally { |
| try { |
| if (metaTable != null) { |
| metaTable.close(); |
| } |
| if (statsTable != null) { |
| statsTable.close(); |
| } |
| } catch (IOException e) {} |
| } |
| } |
| }; |
| Thread t = new Thread(r); |
| t.setDaemon(true); |
| t.start(); |
| |
| if (!enableRebuildIndex) { |
| LOGGER.info("Failure Index Rebuild is skipped by configuration."); |
| return; |
| } |
| // Ensure we only run one of the index rebuilder tasks |
| if (ServerUtil.isKeyInRegion(SYSTEM_CATALOG_KEY, e.getEnvironment().getRegion())) { |
| try { |
| Class.forName(PhoenixDriver.class.getName()); |
| initRebuildIndexConnectionProps(e.getEnvironment().getConfiguration()); |
| // starts index rebuild schedule work |
| BuildIndexScheduleTask task = new BuildIndexScheduleTask(e.getEnvironment()); |
| executor.scheduleWithFixedDelay(task, initialRebuildTaskDelay, rebuildIndexTimeInterval, TimeUnit.MILLISECONDS); |
| } catch (ClassNotFoundException ex) { |
| LOGGER.error("BuildIndexScheduleTask cannot start!", ex); |
| } |
| } |
| } |
| |
| /** |
| * Task runs periodically to build indexes whose INDEX_NEED_PARTIALLY_REBUILD is set true |
| * |
| */ |
| public static class BuildIndexScheduleTask extends TimerTask { |
| RegionCoprocessorEnvironment env; |
| private final long rebuildIndexBatchSize; |
| private final long configuredBatches; |
| private final long indexDisableTimestampThreshold; |
| private final long pendingDisableThreshold; |
| private final ReadOnlyProps props; |
| private final List<String> onlyTheseTables; |
| |
| public BuildIndexScheduleTask(RegionCoprocessorEnvironment env) { |
| this(env,null); |
| } |
| |
| public BuildIndexScheduleTask(RegionCoprocessorEnvironment env, List<String> onlyTheseTables) { |
| this.onlyTheseTables = onlyTheseTables == null ? null : ImmutableList.copyOf(onlyTheseTables); |
| this.env = env; |
| Configuration configuration = env.getConfiguration(); |
| this.rebuildIndexBatchSize = configuration.getLong( |
| QueryServices.INDEX_FAILURE_HANDLING_REBUILD_PERIOD, HConstants.LATEST_TIMESTAMP); |
| this.configuredBatches = configuration.getLong( |
| QueryServices.INDEX_FAILURE_HANDLING_REBUILD_NUMBER_OF_BATCHES_PER_TABLE, 10); |
| this.indexDisableTimestampThreshold = |
| configuration.getLong(QueryServices.INDEX_REBUILD_DISABLE_TIMESTAMP_THRESHOLD, |
| QueryServicesOptions.DEFAULT_INDEX_REBUILD_DISABLE_TIMESTAMP_THRESHOLD); |
| this.pendingDisableThreshold = |
| configuration.getLong(QueryServices.INDEX_PENDING_DISABLE_THRESHOLD, |
| QueryServicesOptions.DEFAULT_INDEX_PENDING_DISABLE_THRESHOLD); |
| this.props = new ReadOnlyProps(env.getConfiguration().iterator()); |
| } |
| |
| public List<PTable> decrementIndexesPendingDisableCount(PhoenixConnection conn, PTable dataPTable, List<PTable> indexes){ |
| List<PTable> indexesIncremented = new ArrayList<>(); |
| for(PTable index :indexes) { |
| try { |
| String indexName = index.getName().getString(); |
| IndexUtil.incrementCounterForIndex(conn, indexName, -PENDING_DISABLE_INACTIVE_STATE_COUNT); |
| indexesIncremented.add(index); |
| }catch(Exception e) { |
| LOGGER.warn("Decrement of -" + PENDING_DISABLE_INACTIVE_STATE_COUNT |
| + " for index :" + index.getName().getString() + "of table: " |
| + dataPTable.getName().getString(), e); |
| } |
| } |
| return indexesIncremented; |
| } |
| @Override |
| public void run() { |
| // FIXME: we should replay the data table Put, as doing a partial index build would only add |
| // the new rows and not delete the previous index value. Also, we should restrict the scan |
| // to only data within this region (as otherwise *every* region will be running this code |
| // separately, all updating the same data. |
| RegionScanner scanner = null; |
| PhoenixConnection conn = null; |
| try { |
| Scan scan = new Scan(); |
| SingleColumnValueFilter filter = new SingleColumnValueFilter(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, |
| PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES, |
| CompareFilter.CompareOp.NOT_EQUAL, PLong.INSTANCE.toBytes(0L)); |
| filter.setFilterIfMissing(true); |
| scan.setFilter(filter); |
| scan.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, |
| PhoenixDatabaseMetaData.TABLE_NAME_BYTES); |
| scan.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, |
| PhoenixDatabaseMetaData.DATA_TABLE_NAME_BYTES); |
| scan.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, |
| PhoenixDatabaseMetaData.INDEX_STATE_BYTES); |
| scan.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, |
| PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES); |
| |
| Map<PTable, List<Pair<PTable,Long>>> dataTableToIndexesMap = null; |
| boolean hasMore = false; |
| List<Cell> results = new ArrayList<Cell>(); |
| scanner = this.env.getRegion().getScanner(scan); |
| |
| do { |
| results.clear(); |
| hasMore = scanner.next(results); |
| if (results.isEmpty()) { |
| LOGGER.debug("Found no indexes with non zero INDEX_DISABLE_TIMESTAMP"); |
| break; |
| } |
| |
| Result r = Result.create(results); |
| byte[] disabledTimeStamp = r.getValue(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, |
| PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES); |
| Cell indexStateCell = r.getColumnLatestCell(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_STATE_BYTES); |
| |
| if (disabledTimeStamp == null || disabledTimeStamp.length == 0) { |
| LOGGER.debug("Null or empty INDEX_DISABLE_TIMESTAMP"); |
| continue; |
| } |
| |
| long indexDisableTimestamp = |
| PLong.INSTANCE.getCodec().decodeLong(disabledTimeStamp, 0, |
| SortOrder.ASC); |
| byte[] dataTable = r.getValue(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, |
| PhoenixDatabaseMetaData.DATA_TABLE_NAME_BYTES); |
| if ((dataTable == null || dataTable.length == 0) || indexStateCell == null) { |
| // data table name can't be empty |
| LOGGER.debug("Null or data table name or index state"); |
| continue; |
| } |
| |
| byte[] indexStateBytes = CellUtil.cloneValue(indexStateCell); |
| byte[][] rowKeyMetaData = new byte[3][]; |
| SchemaUtil.getVarChars(r.getRow(), 3, rowKeyMetaData); |
| byte[] schemaName = rowKeyMetaData[PhoenixDatabaseMetaData.SCHEMA_NAME_INDEX]; |
| byte[] indexTable = rowKeyMetaData[PhoenixDatabaseMetaData.TABLE_NAME_INDEX]; |
| |
| // validity check |
| if (indexTable == null || indexTable.length == 0) { |
| LOGGER.debug("We find IndexTable empty during rebuild scan:" + scan |
| + "so, Index rebuild has been skipped for row=" + r); |
| continue; |
| } |
| |
| String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTable); |
| if (onlyTheseTables != null && !onlyTheseTables.contains(dataTableFullName)) { |
| LOGGER.debug("Could not find " + dataTableFullName + |
| " in " + onlyTheseTables); |
| continue; |
| } |
| |
| if (conn == null) { |
| conn = getRebuildIndexConnection(env.getConfiguration()); |
| dataTableToIndexesMap = Maps.newHashMap(); |
| } |
| PTable dataPTable = PhoenixRuntime.getTableNoCache(conn, dataTableFullName); |
| |
| String indexTableFullName = SchemaUtil.getTableName(schemaName, indexTable); |
| PTable indexPTable = PhoenixRuntime.getTableNoCache(conn, indexTableFullName); |
| // Sanity check in case index was removed from table |
| if (!dataPTable.getIndexes().contains(indexPTable)) { |
| LOGGER.debug(dataTableFullName + " does not contain " + |
| indexPTable.getName().getString()); |
| continue; |
| } |
| |
| PIndexState indexState = PIndexState.fromSerializedValue(indexStateBytes[0]); |
| long elapsedSinceDisable = EnvironmentEdgeManager.currentTimeMillis() - Math.abs(indexDisableTimestamp); |
| |
| // on an index write failure, the server side transitions to PENDING_DISABLE, then the client |
| // retries, and after retries are exhausted, disables the index |
| if (indexState == PIndexState.PENDING_DISABLE) { |
| if (elapsedSinceDisable > pendingDisableThreshold) { |
| // too long in PENDING_DISABLE - client didn't disable the index, so we do it here |
| IndexUtil.updateIndexState(conn, indexTableFullName, PIndexState.DISABLE, indexDisableTimestamp); |
| } |
| continue; |
| } |
| |
| // Only perform relatively expensive check for all regions online when index |
| // is disabled or pending active since that's the state it's placed into when |
| // an index write fails. |
| if ((indexState == PIndexState.DISABLE || indexState == PIndexState.PENDING_ACTIVE) |
| && !MetaDataUtil.tableRegionsOnline(this.env.getConfiguration(), indexPTable)) { |
| LOGGER.debug("Index rebuild has been skipped because not all regions of" + |
| " index table=" + indexPTable.getName() + " are online."); |
| continue; |
| } |
| |
| if (elapsedSinceDisable > indexDisableTimestampThreshold) { |
| /* |
| * It has been too long since the index has been disabled and any future |
| * attempts to reenable it likely will fail. So we are going to mark the |
| * index as disabled and set the index disable timestamp to 0 so that the |
| * rebuild task won't pick up this index again for rebuild. |
| */ |
| try { |
| IndexUtil.updateIndexState(conn, indexTableFullName, PIndexState.DISABLE, 0l); |
| LOGGER.error("Unable to rebuild index " + indexTableFullName |
| + ". Won't attempt again since index disable timestamp is" + |
| " older than current time by " + indexDisableTimestampThreshold |
| + " milliseconds. Manual intervention needed to re-build" + |
| " the index"); |
| } catch (Throwable ex) { |
| LOGGER.error( |
| "Unable to mark index " + indexTableFullName + " as disabled.", ex); |
| } |
| continue; // don't attempt another rebuild irrespective of whether |
| // updateIndexState worked or not |
| } |
| // Allow index to begin incremental maintenance as index is back online and we |
| // cannot transition directly from DISABLED -> ACTIVE |
| if (indexState == PIndexState.DISABLE) { |
| if(IndexUtil.getIndexPendingDisableCount(conn, indexTableFullName) < PENDING_DISABLE_INACTIVE_STATE_COUNT){ |
| // to avoid incrementing again |
| IndexUtil.incrementCounterForIndex(conn, indexTableFullName, PENDING_DISABLE_INACTIVE_STATE_COUNT); |
| } |
| IndexUtil.updateIndexState(conn, indexTableFullName, PIndexState.INACTIVE, null); |
| continue; // Must wait until clients start to do index maintenance again |
| } else if (indexState == PIndexState.PENDING_ACTIVE) { |
| IndexUtil.updateIndexState(conn, indexTableFullName, PIndexState.ACTIVE, null); |
| continue; // Must wait until clients start to do index maintenance again |
| } else if (indexState != PIndexState.INACTIVE && indexState != PIndexState.ACTIVE) { |
| LOGGER.warn("Unexpected index state of " + indexTableFullName + "=" |
| + indexState + ". Skipping partial rebuild attempt."); |
| continue; |
| } |
| long currentTime = EnvironmentEdgeManager.currentTimeMillis(); |
| long forwardOverlapDurationMs = env.getConfiguration().getLong( |
| QueryServices.INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_FORWARD_TIME_ATTRIB, |
| QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_FORWARD_TIME); |
| // Wait until no failures have occurred in at least forwardOverlapDurationMs |
| if (indexStateCell.getTimestamp() + forwardOverlapDurationMs > currentTime) { |
| LOGGER.debug("Still must wait " + (indexStateCell.getTimestamp() + |
| forwardOverlapDurationMs - currentTime) + |
| " before starting rebuild for " + indexTableFullName); |
| continue; // Haven't waited long enough yet |
| } |
| Long upperBoundOfRebuild = indexStateCell.getTimestamp() + forwardOverlapDurationMs; |
| // Pass in upperBoundOfRebuild when setting index state or increasing disable ts |
| // and fail if index timestamp > upperBoundOfRebuild. |
| List<Pair<PTable,Long>> indexesToPartiallyRebuild = dataTableToIndexesMap.get(dataPTable); |
| if (indexesToPartiallyRebuild == null) { |
| indexesToPartiallyRebuild = Lists.newArrayListWithExpectedSize(dataPTable.getIndexes().size()); |
| dataTableToIndexesMap.put(dataPTable, indexesToPartiallyRebuild); |
| } |
| LOGGER.debug("We have found " + indexPTable.getIndexState() + " Index:" + |
| indexPTable.getName() + " on data table:" + dataPTable.getName() + |
| " which failed to be updated at " |
| + indexPTable.getIndexDisableTimestamp()); |
| indexesToPartiallyRebuild.add(new Pair<PTable,Long>(indexPTable,upperBoundOfRebuild)); |
| } while (hasMore); |
| |
| if (dataTableToIndexesMap != null) { |
| long backwardOverlapDurationMs = env.getConfiguration().getLong( |
| QueryServices.INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_BACKWARD_TIME_ATTRIB, |
| env.getConfiguration().getLong(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_TIME_ATTRIB, |
| QueryServicesOptions.DEFAULT_INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_BACKWARD_TIME)); |
| for (Map.Entry<PTable, List<Pair<PTable,Long>>> entry : dataTableToIndexesMap.entrySet()) { |
| PTable dataPTable = entry.getKey(); |
| List<Pair<PTable,Long>> pairs = entry.getValue(); |
| List<PTable> indexesToPartiallyRebuild = Lists.newArrayListWithExpectedSize(pairs.size()); |
| try ( |
| HTableInterface metaTable = env.getTable( |
| SchemaUtil.getPhysicalName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, props))) { |
| long earliestDisableTimestamp = Long.MAX_VALUE; |
| long latestUpperBoundTimestamp = Long.MIN_VALUE; |
| List<IndexMaintainer> maintainers = Lists |
| .newArrayListWithExpectedSize(pairs.size()); |
| int signOfDisableTimeStamp = 0; |
| for (Pair<PTable,Long> pair : pairs) { |
| // We need a way of differentiating the block writes to data table case from |
| // the leave index active case. In either case, we need to know the time stamp |
| // at which writes started failing so we can rebuild from that point. If we |
| // keep the index active *and* have a positive INDEX_DISABLE_TIMESTAMP_BYTES, |
| // then writes to the data table will be blocked (this is client side logic |
| // and we can't change this in a minor release). So we use the sign of the |
| // time stamp to differentiate. |
| PTable index = pair.getFirst(); |
| Long upperBoundTimestamp = pair.getSecond(); |
| long disabledTimeStampVal = index.getIndexDisableTimestamp(); |
| if (disabledTimeStampVal != 0) { |
| if (signOfDisableTimeStamp != 0 && signOfDisableTimeStamp != Long.signum(disabledTimeStampVal)) { |
| LOGGER.warn("Found unexpected mix of signs with " + |
| "INDEX_DISABLE_TIMESTAMP for " + |
| dataPTable.getName().getString() + " with " + |
| indexesToPartiallyRebuild); |
| } |
| signOfDisableTimeStamp = Long.signum(disabledTimeStampVal); |
| disabledTimeStampVal = Math.abs(disabledTimeStampVal); |
| if (disabledTimeStampVal < earliestDisableTimestamp) { |
| earliestDisableTimestamp = disabledTimeStampVal; |
| } |
| |
| indexesToPartiallyRebuild.add(index); |
| maintainers.add(index.getIndexMaintainer(dataPTable, conn)); |
| } |
| if (upperBoundTimestamp > latestUpperBoundTimestamp) { |
| latestUpperBoundTimestamp = upperBoundTimestamp; |
| } |
| } |
| // No indexes are disabled, so skip this table |
| if (earliestDisableTimestamp == Long.MAX_VALUE) { |
| LOGGER.debug("No indexes are disabled so continuing"); |
| continue; |
| } |
| long scanBeginTime = Math.max(0, earliestDisableTimestamp - backwardOverlapDurationMs); |
| long scanEndTime = Math.min(latestUpperBoundTimestamp, |
| getTimestampForBatch(scanBeginTime,batchExecutedPerTableMap.get(dataPTable.getName()))); |
| LOGGER.info("Starting to build " + dataPTable + " indexes " |
| + indexesToPartiallyRebuild + " from timestamp=" + |
| scanBeginTime + " until " + scanEndTime); |
| |
| TableRef tableRef = new TableRef(null, dataPTable, HConstants.LATEST_TIMESTAMP, false); |
| // TODO Need to set high timeout |
| PostDDLCompiler compiler = new PostDDLCompiler(conn); |
| MutationPlan plan = compiler.compile(Collections.singletonList(tableRef), null, null, null, scanEndTime); |
| Scan dataTableScan = IndexManagementUtil.newLocalStateScan(plan.getContext().getScan(), maintainers); |
| |
| dataTableScan.setTimeRange(scanBeginTime, scanEndTime); |
| dataTableScan.setCacheBlocks(false); |
| dataTableScan.setAttribute(BaseScannerRegionObserver.REBUILD_INDEXES, TRUE_BYTES); |
| |
| ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable( |
| ByteUtil.EMPTY_BYTE_ARRAY); |
| IndexMaintainer.serializeAdditional(dataPTable, indexMetaDataPtr, indexesToPartiallyRebuild, |
| conn); |
| byte[] attribValue = ByteUtil.copyKeyBytesIfNecessary(indexMetaDataPtr); |
| dataTableScan.setAttribute(PhoenixIndexCodec.INDEX_PROTO_MD, attribValue); |
| ScanUtil.setClientVersion(dataTableScan, MetaDataProtocol.PHOENIX_VERSION); |
| LOGGER.info("Starting to partially build indexes:" + |
| indexesToPartiallyRebuild + " on data table:" + |
| dataPTable.getName() + " with the earliest disable timestamp:" |
| + earliestDisableTimestamp + " till " |
| + (scanEndTime == HConstants.LATEST_TIMESTAMP ? |
| "LATEST_TIMESTAMP" : scanEndTime)); |
| MutationState mutationState = plan.execute(); |
| long rowCount = mutationState.getUpdateCount(); |
| decrementIndexesPendingDisableCount(conn, dataPTable, indexesToPartiallyRebuild); |
| if (scanEndTime == latestUpperBoundTimestamp) { |
| LOGGER.info("Rebuild completed for all inactive/disabled indexes" + |
| " in data table:" + dataPTable.getName()); |
| } |
| LOGGER.info(" no. of datatable rows read in rebuilding process is " |
| + rowCount); |
| for (PTable indexPTable : indexesToPartiallyRebuild) { |
| String indexTableFullName = SchemaUtil.getTableName( |
| indexPTable.getSchemaName().getString(), |
| indexPTable.getTableName().getString()); |
| try { |
| if (scanEndTime == latestUpperBoundTimestamp) { |
| IndexUtil.updateIndexState(conn, indexTableFullName, PIndexState.ACTIVE, 0L, |
| latestUpperBoundTimestamp); |
| batchExecutedPerTableMap.remove(dataPTable.getName()); |
| LOGGER.info("Making Index:" + indexPTable.getTableName() + |
| " active after rebuilding"); |
| } else { |
| // Increment timestamp so that client sees updated disable timestamp |
| IndexUtil.updateIndexState(conn, indexTableFullName, indexPTable.getIndexState(), |
| scanEndTime * signOfDisableTimeStamp, latestUpperBoundTimestamp); |
| Long noOfBatches = batchExecutedPerTableMap.get(dataPTable.getName()); |
| if (noOfBatches == null) { |
| noOfBatches = 0l; |
| } |
| batchExecutedPerTableMap.put(dataPTable.getName(), ++noOfBatches); |
| LOGGER.info( |
| "During Round-robin build: Successfully updated " + |
| "index disabled timestamp for " |
| + indexTableFullName + " to " + scanEndTime); |
| } |
| } catch (SQLException e) { |
| LOGGER.error("Unable to rebuild " + dataPTable + " index " + |
| indexTableFullName, e); |
| } |
| } |
| } catch (Exception e) { |
| LOGGER.error("Unable to rebuild " + dataPTable + " indexes " + |
| indexesToPartiallyRebuild, e); |
| } |
| } |
| } |
| } catch (Throwable t) { |
| LOGGER.warn("ScheduledBuildIndexTask failed!", t); |
| } finally { |
| if (scanner != null) { |
| try { |
| scanner.close(); |
| } catch (IOException ignored) { |
| LOGGER.debug("ScheduledBuildIndexTask can't close scanner.", ignored); |
| } |
| } |
| if (conn != null) { |
| try { |
| conn.close(); |
| } catch (SQLException ignored) { |
| LOGGER.debug("ScheduledBuildIndexTask can't close connection", ignored); |
| } |
| } |
| } |
| } |
| |
| private long getTimestampForBatch(long disabledTimeStamp, Long noOfBatches) { |
| if (disabledTimeStamp < 0 || rebuildIndexBatchSize > (HConstants.LATEST_TIMESTAMP |
| - disabledTimeStamp)) { return HConstants.LATEST_TIMESTAMP; } |
| long timestampForNextBatch = disabledTimeStamp + rebuildIndexBatchSize; |
| if (timestampForNextBatch < 0 || timestampForNextBatch > EnvironmentEdgeManager.currentTimeMillis() |
| || (noOfBatches != null && noOfBatches > configuredBatches)) { |
| // if timestampForNextBatch cross current time , then we should |
| // build the complete index |
| timestampForNextBatch = HConstants.LATEST_TIMESTAMP; |
| } |
| return timestampForNextBatch; |
| } |
| } |
| |
| @VisibleForTesting |
| public static synchronized void initRebuildIndexConnectionProps(Configuration config) { |
| if (rebuildIndexConnectionProps == null) { |
| Properties props = new Properties(); |
| long indexRebuildQueryTimeoutMs = |
| config.getLong(QueryServices.INDEX_REBUILD_QUERY_TIMEOUT_ATTRIB, |
| QueryServicesOptions.DEFAULT_INDEX_REBUILD_QUERY_TIMEOUT); |
| long indexRebuildRPCTimeoutMs = |
| config.getLong(QueryServices.INDEX_REBUILD_RPC_TIMEOUT_ATTRIB, |
| QueryServicesOptions.DEFAULT_INDEX_REBUILD_RPC_TIMEOUT); |
| long indexRebuildClientScannerTimeOutMs = |
| config.getLong(QueryServices.INDEX_REBUILD_CLIENT_SCANNER_TIMEOUT_ATTRIB, |
| QueryServicesOptions.DEFAULT_INDEX_REBUILD_CLIENT_SCANNER_TIMEOUT); |
| int indexRebuildRpcRetriesCounter = |
| config.getInt(QueryServices.INDEX_REBUILD_RPC_RETRIES_COUNTER, |
| QueryServicesOptions.DEFAULT_INDEX_REBUILD_RPC_RETRIES_COUNTER); |
| // Set various phoenix and hbase level timeouts and rpc retries |
| props.setProperty(QueryServices.THREAD_TIMEOUT_MS_ATTRIB, |
| Long.toString(indexRebuildQueryTimeoutMs)); |
| props.setProperty(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, |
| Long.toString(indexRebuildClientScannerTimeOutMs)); |
| props.setProperty(HConstants.HBASE_RPC_TIMEOUT_KEY, |
| Long.toString(indexRebuildRPCTimeoutMs)); |
| props.setProperty(HConstants.HBASE_CLIENT_RETRIES_NUMBER, |
| Long.toString(indexRebuildRpcRetriesCounter)); |
| // don't run a second index populations upsert select |
| props.setProperty(QueryServices.INDEX_POPULATION_SLEEP_TIME, "0"); |
| rebuildIndexConnectionProps = PropertiesUtil.combineProperties(props, config); |
| } |
| } |
| |
| public static PhoenixConnection getRebuildIndexConnection(Configuration config) |
| throws SQLException, ClassNotFoundException { |
| initRebuildIndexConnectionProps(config); |
| //return QueryUtil.getConnectionOnServer(rebuildIndexConnectionProps, config).unwrap(PhoenixConnection.class); |
| return QueryUtil.getConnectionOnServerWithCustomUrl(rebuildIndexConnectionProps, |
| REBUILD_INDEX_APPEND_TO_URL_STRING).unwrap(PhoenixConnection.class); |
| } |
| } |
