You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2014/08/12 18:46:13 UTC
git commit: PHOENIX-1161 Prevent necessity of clearing region cache
on every access for local indexing
Repository: phoenix
Updated Branches:
refs/heads/4.0 4fa6146b3 -> 04874e69e
PHOENIX-1161 Prevent necessity of clearing region cache on every access for local indexing
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/04874e69
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/04874e69
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/04874e69
Branch: refs/heads/4.0
Commit: 04874e69e3cc598ae86a14b502789a04635fcc15
Parents: 4fa6146
Author: James Taylor <jt...@salesforce.com>
Authored: Tue Aug 12 09:49:53 2014 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Tue Aug 12 09:49:53 2014 -0700
----------------------------------------------------------------------
.../apache/phoenix/end2end/index/LocalIndexIT.java | 17 +++++++----------
.../coprocessor/BaseScannerRegionObserver.java | 9 +++++++--
.../LocalIndexParallelIteratorRegionSplitter.java | 1 -
.../apache/phoenix/iterate/ParallelIterators.java | 15 ++++++++++-----
.../phoenix/query/ConnectionQueryServicesImpl.java | 3 +++
5 files changed, 27 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04874e69/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java
index a291a37..06e4d86 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/LocalIndexIT.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.end2end.index;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -91,7 +92,6 @@ public class LocalIndexIT extends BaseIndexIT {
public void testLocalIndexRoundTrip() throws Exception {
createBaseTable(DATA_TABLE_NAME, null, null);
Connection conn1 = DriverManager.getConnection(getUrl());
- Connection conn2 = DriverManager.getConnection(getUrl());
conn1.createStatement().execute("CREATE LOCAL INDEX " + INDEX_TABLE_NAME + " ON " + DATA_TABLE_NAME + "(v1)");
conn1.createStatement().executeQuery("SELECT * FROM " + DATA_TABLE_FULL_NAME).next();
PTable localIndex = conn1.unwrap(PhoenixConnection.class).getMetaDataCache().getTable(new PTableKey(null,INDEX_TABLE_NAME));
@@ -160,7 +160,7 @@ public class LocalIndexIT extends BaseIndexIT {
assertEquals(IndexRegionSplitPolicy.class.getName(), htd.getValue(HTableDescriptor.SPLIT_POLICY));
HTable userTable = new HTable(admin.getConfiguration(),TableName.valueOf(DATA_TABLE_NAME));
HTable indexTable = new HTable(admin.getConfiguration(),TableName.valueOf(MetaDataUtil.getLocalIndexTableName(DATA_TABLE_NAME)));
- assertEquals("Both user region and index table should have same split keys.", userTable.getStartKeys(), indexTable.getStartKeys());
+ assertArrayEquals("Both user table and index table should have same split keys.", userTable.getStartKeys(), indexTable.getStartKeys());
}
@Test
@@ -593,8 +593,6 @@ public class LocalIndexIT extends BaseIndexIT {
ResultSet rs = conn1.createStatement().executeQuery("SELECT COUNT(*) FROM " + INDEX_TABLE_NAME);
assertTrue(rs.next());
- HBaseAdmin admin = driver.getConnectionQueryServices(getUrl(), TestUtil.TEST_PROPERTIES).getAdmin();
-
String query = "SELECT t_id FROM " + DATA_TABLE_NAME +" where (v1,k3) IN (('z',4),('a',2))";
rs = conn1.createStatement().executeQuery(query);
assertTrue(rs.next());
@@ -647,12 +645,6 @@ public class LocalIndexIT extends BaseIndexIT {
}
assertEquals(4 + i, regionsOfIndexTable.size());
String query = "SELECT t_id,k1,v1 FROM " + DATA_TABLE_NAME;
- rs = conn1.createStatement().executeQuery("EXPLAIN "+query);
- assertEquals(
- "CLIENT PARALLEL " + (4+i) + "-WAY RANGE SCAN OVER "
- + MetaDataUtil.getLocalIndexTableName(DATA_TABLE_NAME)+" [-32768]\n"+
- "CLIENT MERGE SORT",
- QueryUtil.getExplainPlan(rs));
rs = conn1.createStatement().executeQuery(query);
Thread.sleep(1000);
for (int j = 0; j < 26; j++) {
@@ -661,6 +653,11 @@ public class LocalIndexIT extends BaseIndexIT {
assertEquals(25-j, rs.getInt("k1"));
assertEquals(strings[j], rs.getString("V1"));
}
+ rs = conn1.createStatement().executeQuery("EXPLAIN " + query);
+ assertEquals(
+ "CLIENT PARALLEL " + (4 + i) + "-WAY RANGE SCAN OVER "
+ + MetaDataUtil.getLocalIndexTableName(DATA_TABLE_NAME) + " [-32768]\n"
+ + "CLIENT MERGE SORT", QueryUtil.getExplainPlan(rs));
query = "SELECT t_id,k1,k3 FROM " + DATA_TABLE_NAME;
rs = conn1.createStatement().executeQuery("EXPLAIN "+query);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04874e69/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
index 730f2a1..ac8d3a2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
@@ -57,6 +57,7 @@ abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
public static final String LOCAL_INDEX_JOIN_SCHEMA = "_LocalIndexJoinSchema";
public static final String DATA_TABLE_COLUMNS_TO_JOIN = "_DataTableColumnsToJoin";
public static final String VIEW_CONSTANTS = "_ViewConstants";
+ public static final String EXPECTED_UPPER_REGION_KEY = "_ExpectedUpperRegionKey";
/** Exposed for testing */
public static final String SCANNER_OPENED_TRACE_INFO = "Scanner opened on server";
@@ -83,8 +84,12 @@ abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
byte[] upperExclusiveScanKey = scan.getStopRow();
byte[] lowerInclusiveRegionKey = region.getStartKey();
byte[] upperExclusiveRegionKey = region.getEndKey();
- if (Bytes.compareTo(lowerInclusiveScanKey, lowerInclusiveRegionKey) < 0 ||
- (Bytes.compareTo(upperExclusiveScanKey, upperExclusiveRegionKey) > 0 && upperExclusiveRegionKey.length != 0) ) {
+ byte[] expectedUpperRegionKey = scan.getAttribute(EXPECTED_UPPER_REGION_KEY);
+ if ( (expectedUpperRegionKey != null && // local index check
+ Bytes.compareTo(upperExclusiveRegionKey, expectedUpperRegionKey) != 0)
+ || (expectedUpperRegionKey == null && // non local index check
+ ( Bytes.compareTo(lowerInclusiveScanKey, lowerInclusiveRegionKey) < 0 ||
+ ( Bytes.compareTo(upperExclusiveScanKey, upperExclusiveRegionKey) > 0 && upperExclusiveRegionKey.length != 0) ) ) ) {
@SuppressWarnings("deprecation")
Exception cause = new StaleRegionBoundaryCacheException(region.getRegionInfo().getTableName());
throw new DoNotRetryIOException(cause.getMessage(), cause);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04874e69/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java
index 0ccd738..c3a38d5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/LocalIndexParallelIteratorRegionSplitter.java
@@ -37,7 +37,6 @@ public class LocalIndexParallelIteratorRegionSplitter extends DefaultParallelIte
@Override
protected List<HRegionLocation> getAllRegions() throws SQLException {
- context.getConnection().getQueryServices().clearTableRegionCache(tableRef.getTable().getPhysicalName().getBytes());
return context.getConnection().getQueryServices().getAllTableRegions(tableRef.getTable().getPhysicalName().getBytes());
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04874e69/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
index 7600e47..dc8fa67 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
@@ -17,6 +17,8 @@
*/
package org.apache.phoenix.iterate;
+import static org.apache.phoenix.coprocessor.BaseScannerRegionObserver.EXPECTED_UPPER_REGION_KEY;
+
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
@@ -272,7 +274,7 @@ public class ParallelIterators extends ExplainTable implements ResultIterators {
}
});
boolean clearedCache = false;
- byte[] tableName = tableRef.getTable().getName().getBytes();
+ byte[] tableName = tableRef.getTable().getPhysicalName().getBytes();
for (Pair<KeyRange,Future<PeekingResultIterator>> future : futures) {
try {
PeekingResultIterator iterator = future.getSecond().get(timeoutMs, TimeUnit.MILLISECONDS);
@@ -324,7 +326,7 @@ public class ParallelIterators extends ExplainTable implements ResultIterators {
final ConnectionQueryServices services = context.getConnection().getQueryServices();
ExecutorService executor = services.getExecutor();
final boolean localIndex = this.tableRef.getTable().getType() == PTableType.INDEX && this.tableRef.getTable().getIndexType() == IndexType.LOCAL;
- for (KeyRange split : splits) {
+ for (final KeyRange split : splits) {
final Scan splitScan = ScanUtil.newScan(context.getScan());
// Intersect with existing start/stop key if the table is salted
// If not salted, we've already intersected it. If salted, we need
@@ -333,11 +335,14 @@ public class ParallelIterators extends ExplainTable implements ResultIterators {
if (tableRef.getTable().getBucketNum() != null) {
KeyRange minMaxRange = context.getMinMaxRange();
if (minMaxRange != null) {
- // Add salt byte based on current split, as minMaxRange won't have it
- minMaxRange = SaltingUtil.addSaltByte(split.getLowerRange(), minMaxRange);
- split = split.intersect(minMaxRange);
+ KeyRange splitScanRange = KeyRange.getKeyRange(splitScan.getStartRow(), splitScan.getStopRow());
+ splitScanRange = splitScanRange.intersect(SaltingUtil.addSaltByte(split.getLowerRange(), minMaxRange));
+ splitScan.setStartRow(splitScanRange.getLowerRange());
+ splitScan.setStopRow(splitScanRange.getUpperRange());
}
} else if (localIndex) {
+ // Used to detect stale region boundary information on server side
+ splitScan.setAttribute(EXPECTED_UPPER_REGION_KEY, split.getUpperRange());
if (splitScan.getStartRow().length != 0 || splitScan.getStopRow().length != 0) {
SaltingUtil.addRegionStartKeyToScanStartAndStopRows(split.getLowerRange(),split.getUpperRange(),
splitScan);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/04874e69/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 4965813..74e05da 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -1031,6 +1031,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if (dropMetadata) {
admin.disableTable(physicalIndexName);
admin.deleteTable(physicalIndexName);
+ clearTableRegionCache(physicalIndexName);
wasDeleted = true;
}
}
@@ -1064,6 +1065,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if (dropMetadata) {
admin.disableTable(physicalIndexName);
admin.deleteTable(physicalIndexName);
+ clearTableRegionCache(physicalIndexName);
wasDeleted = true;
}
}
@@ -1246,6 +1248,7 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
if ( admin.tableExists(tableName) ) {
admin.disableTable(tableName);
admin.deleteTable(tableName);
+ clearTableRegionCache(tableName);
}
}
}