phoenix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jamestay...@apache.org
Subject [3/3] phoenix git commit: PHOENIX-3953 Clear INDEX_DISABLED_TIMESTAMP and disable index on compaction (addendum 2)
Date Thu, 07 Sep 2017 18:35:56 GMT
PHOENIX-3953 Clear INDEX_DISABLED_TIMESTAMP and disable index on compaction (addendum 2)


Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/5cf07c4c
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5cf07c4c
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5cf07c4c

Branch: refs/heads/4.x-HBase-1.2
Commit: 5cf07c4ce64174241e0c311c6f9e1905374aaeca
Parents: aea6106
Author: James Taylor <jtaylor@salesforce.com>
Authored: Thu Sep 7 11:26:47 2017 -0700
Committer: James Taylor <jtaylor@salesforce.com>
Committed: Thu Sep 7 11:34:53 2017 -0700

----------------------------------------------------------------------
 .../UngroupedAggregateRegionObserver.java       | 62 ++++++++++++++-
 .../org/apache/phoenix/hbase/index/Indexer.java | 42 +++++-----
 .../stats/DefaultStatisticsCollector.java       | 83 ++++++--------------
 .../schema/stats/NoOpStatisticsCollector.java   |  2 +-
 .../schema/stats/StatisticsCollector.java       |  2 +-
 5 files changed, 111 insertions(+), 80 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cf07c4c/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index 31c83e4..a61f502 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -55,9 +55,12 @@ import org.apache.hadoop.hbase.NamespaceDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.HTableInterface;
 import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
@@ -69,11 +72,14 @@ import org.apache.hadoop.hbase.regionserver.Region;
 import org.apache.hadoop.hbase.regionserver.RegionScanner;
 import org.apache.hadoop.hbase.regionserver.ScanType;
 import org.apache.hadoop.hbase.regionserver.Store;
+import org.apache.hadoop.hbase.regionserver.StoreFile;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.hadoop.io.WritableUtils;
 import org.apache.phoenix.cache.ServerCacheClient;
+import org.apache.phoenix.coprocessor.MetaDataProtocol.MutationCode;
 import org.apache.phoenix.coprocessor.generated.PTableProtos;
 import org.apache.phoenix.exception.DataExceedsCapacityException;
 import org.apache.phoenix.execute.TupleProjector;
@@ -89,11 +95,13 @@ import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
 import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
 import org.apache.phoenix.index.IndexMaintainer;
 import org.apache.phoenix.index.PhoenixIndexCodec;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
 import org.apache.phoenix.join.HashJoinInfo;
 import org.apache.phoenix.query.QueryConstants;
 import org.apache.phoenix.query.QueryServicesOptions;
 import org.apache.phoenix.schema.ColumnFamilyNotFoundException;
 import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PIndexState;
 import org.apache.phoenix.schema.PRow;
 import org.apache.phoenix.schema.PTable;
 import org.apache.phoenix.schema.PTableImpl;
@@ -899,6 +907,58 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
         });
     }
 
+    @Override
+    public void postCompact(final ObserverContext<RegionCoprocessorEnvironment> e,
final Store store,
+            final StoreFile resultFile, CompactionRequest request) throws IOException {
+        // If we're compacting all files, then delete markers are removed
+        // and we must permanently disable an index that needs to be
+        // partially rebuild because we're potentially losing the information
+        // we need to successfully rebuilt it.
+        if (request.isAllFiles() || request.isMajor()) {
+            // Compaction and split upcalls run with the effective user context of the requesting
user.
+            // This will lead to failure of cross cluster RPC if the effective user is not
+            // the login user. Switch to the login user context to ensure we have the expected
+            // security context.
+            User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
+                @Override
+                public Void run() throws Exception {
+                    MutationCode mutationCode = null;
+                    long disableIndexTimestamp = 0;
+                    
+                    try (HTableInterface htable = e.getEnvironment().getTable(
+                                SchemaUtil.getPhysicalTableName(
+                                        PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES,
+                                        e.getEnvironment().getConfiguration()))) {
+                        String tableName = e.getEnvironment().getRegion().getRegionInfo().getTable().getNameAsString();
+                        // FIXME: if this is an index on a view, we won't find a row for
it in SYSTEM.CATALOG
+                        // Instead, we need to disable all indexes on the view.
+                        byte[] tableKey = SchemaUtil.getTableKeyFromFullName(tableName);
+                        Get get = new Get(tableKey);
+                        get.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES);
+                        Result result = htable.get(get);
+                        if (!result.isEmpty()) {
+                            Cell cell = result.listCells().get(0);
+                            if (cell.getValueLength() > 0) {
+                                disableIndexTimestamp = PLong.INSTANCE.getCodec().decodeLong(cell.getValueArray(),
cell.getValueOffset(), SortOrder.getDefault());
+                                if (disableIndexTimestamp != 0) {
+                                    mutationCode = IndexUtil.updateIndexState(tableKey, 0L,
htable, PIndexState.DISABLE).getMutationCode();
+                                }
+                            }
+                        }
+                    } catch (Throwable t) { // log, but swallow exception as we don't want
to impact compaction
+                        logger.warn("Potential failure to permanently disable index during
compaction " +  e.getEnvironment().getRegionInfo().getTable().getNameAsString(), t);
+                    } finally {
+                        if (disableIndexTimestamp != 0 && mutationCode != MutationCode.TABLE_ALREADY_EXISTS
&& mutationCode != MutationCode.TABLE_NOT_FOUND) {
+                            logger.warn("Attempt to permanently disable index " + e.getEnvironment().getRegionInfo().getTable().getNameAsString()
+ 
+                                    " during compaction" + (mutationCode == null ? "" : "
failed with code = " + mutationCode));
+                        }
+                    }
+                    return null;
+                }
+            });
+        }
+    }
+
     private static PTable deserializeTable(byte[] b) {
         try {
             PTableProtos.PTable ptableProto = PTableProtos.PTable.parseFrom(b);
@@ -1115,7 +1175,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
             long rowCount = 0;
             try {
                 if (!compactionRunning) {
-                    stats.init(false);
+                    stats.init();
                     synchronized (innerScanner) {
                         do {
                             List<Cell> results = new ArrayList<Cell>();

http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cf07c4c/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
index 8072fba..4273eb1 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
@@ -62,6 +62,8 @@ import org.apache.hadoop.hbase.regionserver.OperationStatus;
 import org.apache.hadoop.hbase.regionserver.Region;
 import org.apache.hadoop.hbase.regionserver.ScanType;
 import org.apache.hadoop.hbase.regionserver.Store;
+import org.apache.hadoop.hbase.regionserver.StoreFile;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
 import org.apache.hadoop.hbase.security.User;
@@ -845,21 +847,25 @@ public class Indexer extends BaseRegionObserver {
   }
   
   @Override
-  public InternalScanner preCompact(final ObserverContext<RegionCoprocessorEnvironment>
c, final Store store,
-          final InternalScanner scanner, final ScanType scanType) throws IOException {
-      // Compaction and split upcalls run with the effective user context of the requesting
user.
-      // This will lead to failure of cross cluster RPC if the effective user is not
-      // the login user. Switch to the login user context to ensure we have the expected
-      // security context.
-      return User.runAsLoginUser(new PrivilegedExceptionAction<InternalScanner>() {
-          @Override
-          public InternalScanner run() throws Exception {
-              InternalScanner internalScanner = scanner;
-              if (scanType.equals(ScanType.COMPACT_DROP_DELETES)) {
+  public void postCompact(final ObserverContext<RegionCoprocessorEnvironment> c, final
Store store,
+          final StoreFile resultFile, CompactionRequest request) throws IOException {
+      // If we're compacting all files, then delete markers are removed
+      // and we must permanently disable an index that needs to be
+      // partially rebuild because we're potentially losing the information
+      // we need to successfully rebuilt it.
+      if (request.isAllFiles() || request.isMajor()) {
+          // Compaction and split upcalls run with the effective user context of the requesting
user.
+          // This will lead to failure of cross cluster RPC if the effective user is not
+          // the login user. Switch to the login user context to ensure we have the expected
+          // security context.
+          User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
+              @Override
+              public Void run() throws Exception {
                   String fullTableName = c.getEnvironment().getRegion().getRegionInfo().getTable().getNameAsString();
                   try {
                       PhoenixConnection conn =  QueryUtil.getConnectionOnServer(c.getEnvironment().getConfiguration()).unwrap(PhoenixConnection.class);
                       PTable table = PhoenixRuntime.getTableNoCache(conn, fullTableName);
+                      // FIXME: we may need to recurse into children of this table too
                       for (PTable index : table.getIndexes()) {
                           if (index.getIndexDisableTimestamp() != 0) {
                               try {
@@ -871,15 +877,15 @@ public class Indexer extends BaseRegionObserver {
                       }
                   } catch (Exception e) {
                       // If we can't reach the stats table, don't interrupt the normal
-                    // compaction operation, just log a warning.
-                    if (LOG.isWarnEnabled()) {
-                        LOG.warn("Unable to permanently disable indexes being partially rebuild
for " + fullTableName, e);
-                    }
+                      // compaction operation, just log a warning.
+                      if (LOG.isWarnEnabled()) {
+                          LOG.warn("Unable to permanently disable indexes being partially
rebuild for " + fullTableName, e);
+                      }
                   }
+                  return null;
               }
-              return internalScanner;
-          }
-      });
+          });
+      }
   }
 }
 

http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cf07c4c/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
index 61a6fa2..b8ba759 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
@@ -29,7 +29,6 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.KeyValueUtil;
-import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.HTableInterface;
 import org.apache.hadoop.hbase.client.Mutation;
@@ -42,16 +41,13 @@ import org.apache.hadoop.hbase.regionserver.Region;
 import org.apache.hadoop.hbase.regionserver.Store;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.phoenix.coprocessor.MetaDataProtocol;
-import org.apache.phoenix.coprocessor.MetaDataProtocol.MutationCode;
 import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
 import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
 import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.query.QueryServicesOptions;
-import org.apache.phoenix.schema.PIndexState;
 import org.apache.phoenix.schema.SortOrder;
 import org.apache.phoenix.schema.types.PInteger;
 import org.apache.phoenix.schema.types.PLong;
-import org.apache.phoenix.util.IndexUtil;
 import org.apache.phoenix.util.MetaDataUtil;
 import org.apache.phoenix.util.SchemaUtil;
 import org.apache.phoenix.util.TimeKeeper;
@@ -113,10 +109,9 @@ class DefaultStatisticsCollector implements StatisticsCollector {
         }
     }
     
-    private void initGuidepostDepth(boolean isMajorCompaction) throws IOException {
+    private void initGuidepostDepth() throws IOException {
         // First check is if guidepost info set on statement itself
-        boolean guidepostOnStatement = guidePostPerRegionBytes != null || guidePostWidthBytes
!= null;
-        if (guidepostOnStatement) {
+        if (guidePostPerRegionBytes != null || guidePostWidthBytes != null) {
             int guidepostPerRegion = 0;
             long guidepostWidth = QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_WIDTH_BYTES;
             if (guidePostPerRegionBytes != null) {
@@ -127,48 +122,20 @@ class DefaultStatisticsCollector implements StatisticsCollector {
             }
             this.guidePostDepth = StatisticsUtil.getGuidePostDepth(guidepostPerRegion, guidepostWidth,
                     env.getRegion().getTableDesc());
-        }
-        
-        if (!guidepostOnStatement || isMajorCompaction) {
+        } else {
             long guidepostWidth = -1;
             HTableInterface htable = null;
             try {
-                // Next check for GUIDE_POST_WIDTH and INDEX_DISABLE_TIMESTAMP on table
-                TableName htableName = SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES,
env.getConfiguration());
-                htable = env.getTable(htableName);
+                // Next check for GUIDE_POST_WIDTH on table
+                htable = env.getTable(
+                        SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES,
env.getConfiguration()));
                 Get get = new Get(ptableKey);
                 get.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH_BYTES);
-                get.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES);
                 Result result = htable.get(get);
                 if (!result.isEmpty()) {
-                    Cell gpwCell = result.getColumnLatestCell(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH_BYTES);
-                    if (gpwCell != null) {
-                        guidepostWidth = PLong.INSTANCE.getCodec().decodeLong(gpwCell.getValueArray(),
gpwCell.getValueOffset(), SortOrder.getDefault());
-                    }
-                    if (isMajorCompaction) {
-                        Cell idtsCell = result.getColumnLatestCell(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP_BYTES);
-                        if (idtsCell != null) {
-                            long indexDisableTimestamp = PLong.INSTANCE.getCodec().decodeLong(idtsCell.getValueArray(),
idtsCell.getValueOffset(), SortOrder.getDefault());
-                            // If we have a non zero value for INDEX_DISABLE_TIMESTAMP, that
means that our global mutable
-                            // secondary index needs to be partially rebuilt. If we're  compacting,
though, we may cleanup
-                            // the delete markers of an index *before* the puts for the same
row occur during replay. At
-                            // this point the partially index rebuild would leave the index
out of sync with the data
-                            // table. In that case, it's better to just permanently disable
the index and force it to be
-                            // manually rebuilt
-                            if (indexDisableTimestamp != 0) {
-                                MutationCode mutationCode = IndexUtil.updateIndexState(ptableKey,
0L, htable, PIndexState.DISABLE).getMutationCode();
-                                if (mutationCode != MutationCode.TABLE_ALREADY_EXISTS &&
mutationCode != MutationCode.TABLE_NOT_FOUND) {
-                                    LOG.warn("Attempt to permanently disable index " + env.getRegionInfo().getTable().getNameAsString()
+ 
-                                            " during compaction failed with code = " + mutationCode);
-                                }
-                            }
-                        }
-                    }
+                    Cell cell = result.listCells().get(0);
+                    guidepostWidth = PLong.INSTANCE.getCodec().decodeLong(cell.getValueArray(),
cell.getValueOffset(), SortOrder.getDefault());
                 }
-            } catch (IOException e) {
-                throw e;
-            } catch (Throwable t) {
-                throw new IOException(t);
             } finally {
                 if (htable != null) {
                     try {
@@ -178,21 +145,19 @@ class DefaultStatisticsCollector implements StatisticsCollector {
                     }
                 }
             }
-            if (!guidepostOnStatement) {
-                if (guidepostWidth >= 0) {
-                    this.guidePostDepth = guidepostWidth;
-                } else {
-                    // Last use global config value
-                    Configuration config = env.getConfiguration();
-                    this.guidePostDepth = StatisticsUtil.getGuidePostDepth(
-                            config.getInt(
-                                QueryServices.STATS_GUIDEPOST_PER_REGION_ATTRIB,
-                                QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_PER_REGION),
-                            config.getLong(
-                                QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB,
-                                QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_WIDTH_BYTES),
-                            env.getRegion().getTableDesc());
-                }
+            if (guidepostWidth >= 0) {
+                this.guidePostDepth = guidepostWidth;
+            } else {
+                // Last use global config value
+                Configuration config = env.getConfiguration();
+                this.guidePostDepth = StatisticsUtil.getGuidePostDepth(
+                        config.getInt(
+                            QueryServices.STATS_GUIDEPOST_PER_REGION_ATTRIB,
+                            QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_PER_REGION),
+                        config.getLong(
+                            QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB,
+                            QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_WIDTH_BYTES),
+                        env.getRegion().getTableDesc());
             }
         }
     }
@@ -350,13 +315,13 @@ class DefaultStatisticsCollector implements StatisticsCollector {
         StatisticsScanner scanner = new StatisticsScanner(this, statsWriter, env, internalScan,
family);
         // We need to initialize the scanner synchronously and potentially perform a cross
region Get
         // in order to use the correct guide posts width for the table being compacted.
-        init(true);
+        init();
         return scanner;
     }
 
     @Override
-    public void init(boolean isMajorCompaction) throws IOException {
-        initGuidepostDepth(isMajorCompaction);
+    public void init() throws IOException {
+        initGuidepostDepth();
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cf07c4c/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java
b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java
index a13a722..74d1710 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java
@@ -61,7 +61,7 @@ public class NoOpStatisticsCollector implements StatisticsCollector {
     }
 
     @Override 
-    public void init(boolean isMajorCompaction) {
+    public void init() {
         // No-op
     }
 

http://git-wip-us.apache.org/repos/asf/phoenix/blob/5cf07c4c/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java
b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java
index 9550469..60e83a8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java
@@ -64,7 +64,7 @@ public interface StatisticsCollector extends Closeable {
      * Called before beginning the collection of statistics through {@link #collectStatistics(List)}
      * @throws IOException 
      */
-    void init(boolean isMajorCompaction) throws IOException;
+    void init() throws IOException;
 
     /**
      * Retrieve the calculated guide post info for the given column family.


Mime
View raw message