| /* |
| * 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.iterate; |
| |
| import static org.apache.phoenix.coprocessor.BaseScannerRegionObserver.SCAN_ACTUAL_START_ROW; |
| import static org.apache.phoenix.coprocessor.BaseScannerRegionObserver.SCAN_START_ROW_SUFFIX; |
| import static org.apache.phoenix.iterate.TableResultIterator.RenewLeaseStatus.CLOSED; |
| import static org.apache.phoenix.iterate.TableResultIterator.RenewLeaseStatus.LOCK_NOT_ACQUIRED; |
| import static org.apache.phoenix.iterate.TableResultIterator.RenewLeaseStatus.NOT_RENEWED; |
| import static org.apache.phoenix.iterate.TableResultIterator.RenewLeaseStatus.NOT_SUPPORTED; |
| import static org.apache.phoenix.iterate.TableResultIterator.RenewLeaseStatus.RENEWED; |
| import static org.apache.phoenix.iterate.TableResultIterator.RenewLeaseStatus.THRESHOLD_NOT_REACHED; |
| import static org.apache.phoenix.iterate.TableResultIterator.RenewLeaseStatus.UNINITIALIZED; |
| |
| import java.io.IOException; |
| import java.sql.SQLException; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReentrantLock; |
| |
| import javax.annotation.concurrent.GuardedBy; |
| |
| import org.apache.hadoop.hbase.client.AbstractClientScanner; |
| import org.apache.hadoop.hbase.client.Scan; |
| import org.apache.hadoop.hbase.client.Table; |
| import org.apache.hadoop.hbase.io.ImmutableBytesWritable; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.phoenix.cache.ServerCacheClient.ServerCache; |
| import org.apache.phoenix.compile.QueryPlan; |
| import org.apache.phoenix.coprocessor.HashJoinCacheNotFoundException; |
| import org.apache.phoenix.execute.BaseQueryPlan; |
| import org.apache.phoenix.execute.MutationState; |
| import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; |
| import org.apache.phoenix.join.HashCacheClient; |
| import org.apache.phoenix.monitoring.ScanMetricsHolder; |
| import org.apache.phoenix.query.QueryConstants; |
| import org.apache.phoenix.schema.PTable; |
| import org.apache.phoenix.schema.tuple.Tuple; |
| import org.apache.phoenix.util.ByteUtil; |
| import org.apache.phoenix.util.Closeables; |
| import org.apache.phoenix.util.EnvironmentEdgeManager; |
| import org.apache.phoenix.util.IndexUtil; |
| import org.apache.phoenix.util.ScanUtil; |
| import org.apache.phoenix.util.ServerUtil; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| |
| |
| /** |
| * |
| * Wrapper for ResultScanner creation that closes HTableInterface |
| * when ResultScanner is closed. |
| * |
| * |
| * @since 0.1 |
| */ |
| public class TableResultIterator implements ResultIterator { |
| private final Scan scan; |
| private final Table htable; |
| private final ScanMetricsHolder scanMetricsHolder; |
| private static final ResultIterator UNINITIALIZED_SCANNER = ResultIterator.EMPTY_ITERATOR; |
| private final long renewLeaseThreshold; |
| private final QueryPlan plan; |
| private final ParallelScanGrouper scanGrouper; |
| private static final Logger LOGGER = LoggerFactory.getLogger(TableResultIterator.class); |
| private Tuple lastTuple = null; |
| private ImmutableBytesWritable ptr = new ImmutableBytesWritable(); |
| @GuardedBy("renewLeaseLock") |
| private ResultIterator scanIterator; |
| |
| @GuardedBy("renewLeaseLock") |
| private boolean closed = false; |
| |
| @GuardedBy("renewLeaseLock") |
| private long renewLeaseTime = 0; |
| |
| private final Lock renewLeaseLock = new ReentrantLock(); |
| |
| private int retry; |
| private Map<ImmutableBytesPtr,ServerCache> caches; |
| private HashCacheClient hashCacheClient; |
| |
| @VisibleForTesting // Exposed for testing. DON'T USE ANYWHERE ELSE! |
| TableResultIterator() { |
| this.scanMetricsHolder = null; |
| this.renewLeaseThreshold = 0; |
| this.htable = null; |
| this.scan = null; |
| this.plan = null; |
| this.scanGrouper = null; |
| this.caches = null; |
| this.retry = 0; |
| } |
| |
| public static enum RenewLeaseStatus { |
| RENEWED, NOT_RENEWED, CLOSED, UNINITIALIZED, THRESHOLD_NOT_REACHED, LOCK_NOT_ACQUIRED, NOT_SUPPORTED |
| }; |
| |
| public TableResultIterator(MutationState mutationState, Scan scan, ScanMetricsHolder scanMetricsHolder, |
| long renewLeaseThreshold, QueryPlan plan, ParallelScanGrouper scanGrouper) throws SQLException { |
| this(mutationState, scan, scanMetricsHolder, renewLeaseThreshold, plan, scanGrouper, null); |
| } |
| |
| public TableResultIterator(MutationState mutationState, Scan scan, ScanMetricsHolder scanMetricsHolder, |
| long renewLeaseThreshold, QueryPlan plan, ParallelScanGrouper scanGrouper,Map<ImmutableBytesPtr,ServerCache> caches) throws SQLException { |
| this.scan = scan; |
| this.scanMetricsHolder = scanMetricsHolder; |
| this.plan = plan; |
| PTable table = plan.getTableRef().getTable(); |
| htable = mutationState.getHTable(table); |
| this.scanIterator = UNINITIALIZED_SCANNER; |
| this.renewLeaseThreshold = renewLeaseThreshold; |
| this.scanGrouper = scanGrouper; |
| this.hashCacheClient = new HashCacheClient(plan.getContext().getConnection()); |
| this.caches = caches; |
| this.retry=plan.getContext().getConnection().getQueryServices().getProps() |
| .getInt(QueryConstants.HASH_JOIN_CACHE_RETRIES, QueryConstants.DEFAULT_HASH_JOIN_CACHE_RETRIES); |
| IndexUtil.setScanAttributesForIndexReadRepair(scan, table, plan.getContext().getConnection()); |
| } |
| |
| @Override |
| public void close() throws SQLException { |
| try { |
| renewLeaseLock.lock(); |
| closed = true; // ok to say closed even if the below code throws an exception |
| try { |
| scanIterator.close(); |
| } finally { |
| try { |
| scanIterator = UNINITIALIZED_SCANNER; |
| htable.close(); |
| } catch (IOException e) { |
| throw ServerUtil.parseServerException(e); |
| } |
| } |
| } finally { |
| renewLeaseLock.unlock(); |
| } |
| |
| } |
| |
| @Override |
| public Tuple next() throws SQLException { |
| try { |
| renewLeaseLock.lock(); |
| initScanner(); |
| try { |
| lastTuple = scanIterator.next(); |
| if (lastTuple != null) { |
| ImmutableBytesWritable ptr = new ImmutableBytesWritable(); |
| lastTuple.getKey(ptr); |
| } |
| } catch (SQLException e) { |
| try { |
| throw ServerUtil.parseServerException(e); |
| } catch(HashJoinCacheNotFoundException e1) { |
| if(ScanUtil.isNonAggregateScan(scan) && plan.getContext().getAggregationManager().isEmpty()) { |
| // For non aggregate queries if we get stale region boundary exception we can |
| // continue scanning from the next value of lasted fetched result. |
| Scan newScan = ScanUtil.newScan(scan); |
| newScan.withStartRow(newScan.getAttribute(SCAN_ACTUAL_START_ROW)); |
| if(lastTuple != null) { |
| lastTuple.getKey(ptr); |
| byte[] startRowSuffix = ByteUtil.copyKeyBytesIfNecessary(ptr); |
| if(ScanUtil.isLocalIndex(newScan)) { |
| // If we just set scan start row suffix then server side we prepare |
| // actual scan boundaries by prefixing the region start key. |
| newScan.setAttribute(SCAN_START_ROW_SUFFIX, ByteUtil.nextKey(startRowSuffix)); |
| } else { |
| newScan.withStartRow(ByteUtil.nextKey(startRowSuffix)); |
| } |
| } |
| plan.getContext().getConnection().getQueryServices().clearTableRegionCache(htable.getName()); |
| LOGGER.debug( |
| "Retrying when Hash Join cache is not found on the server ,by sending the cache again"); |
| if (retry <= 0) { |
| throw e1; |
| } |
| Long cacheId = e1.getCacheId(); |
| retry--; |
| try { |
| ServerCache cache = caches == null ? null : |
| caches.get(new ImmutableBytesPtr(Bytes.toBytes(cacheId))); |
| if (!hashCacheClient.addHashCacheToServer(newScan.getStartRow(), |
| cache, plan.getTableRef().getTable())) { |
| throw e1; |
| } |
| this.scanIterator = ((BaseQueryPlan) plan).iterator(caches, scanGrouper, newScan); |
| |
| } catch (Exception ex) { |
| throw ServerUtil.parseServerException(ex); |
| } |
| lastTuple = scanIterator.next(); |
| } else { |
| throw e; |
| } |
| } |
| } |
| return lastTuple; |
| } finally { |
| renewLeaseLock.unlock(); |
| } |
| } |
| |
| public void initScanner() throws SQLException { |
| try { |
| renewLeaseLock.lock(); |
| if (closed) { |
| return; |
| } |
| ResultIterator delegate = this.scanIterator; |
| if (delegate == UNINITIALIZED_SCANNER) { |
| try { |
| this.scanIterator = |
| new ScanningResultIterator(htable.getScanner(scan), scan, scanMetricsHolder); |
| } catch (IOException e) { |
| Closeables.closeQuietly(htable); |
| throw ServerUtil.parseServerException(e); |
| } |
| } |
| } finally { |
| renewLeaseLock.unlock(); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| return "TableResultIterator [htable=" + htable + ", scan=" + scan + "]"; |
| } |
| |
| public RenewLeaseStatus renewLease() { |
| boolean lockAcquired = false; |
| try { |
| lockAcquired = renewLeaseLock.tryLock(); |
| if (lockAcquired) { |
| if (closed) { |
| return CLOSED; |
| } |
| if (scanIterator == UNINITIALIZED_SCANNER) { |
| return UNINITIALIZED; |
| } |
| long delay = now() - renewLeaseTime; |
| if (delay < renewLeaseThreshold) { |
| return THRESHOLD_NOT_REACHED; |
| } |
| if (scanIterator instanceof ScanningResultIterator |
| && ((ScanningResultIterator)scanIterator).getScanner() instanceof AbstractClientScanner) { |
| // Need this explicit cast because HBase's ResultScanner doesn't have this method exposed. |
| boolean leaseRenewed = ((AbstractClientScanner)((ScanningResultIterator)scanIterator).getScanner()).renewLease(); |
| if (leaseRenewed) { |
| renewLeaseTime = now(); |
| return RENEWED; |
| } else { |
| return NOT_RENEWED; |
| } |
| } else { |
| return NOT_SUPPORTED; |
| } |
| } |
| return LOCK_NOT_ACQUIRED; |
| } |
| finally { |
| if (lockAcquired) { |
| renewLeaseLock.unlock(); |
| } |
| } |
| } |
| |
| private static long now() { |
| return EnvironmentEdgeManager.currentTimeMillis(); |
| } |
| |
| @Override |
| public void explain(List<String> planSteps) { |
| scanIterator.explain(planSteps); |
| } |
| |
| } |