You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2021/09/13 03:32:08 UTC
[hbase] branch HBASE-26067 updated (9300708 -> d76ba96)
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a change to branch HBASE-26067
in repository https://gitbox.apache.org/repos/asf/hbase.git.
omit 9300708 HBASE-26246 Persist the StoreFileTracker configurations to TableDescriptor when creating table (#3666)
omit 99cb945 HBASE-26224 Introduce a MigrationStoreFileTracker to support migrating from different store file tracker implementations (#3656)
omit 8205d81 HBASE-26079 Use StoreFileTracker when splitting and merging (#3617)
omit 21ae850 HBASE-25988 Store the store file list by a file (#3578)
omit 76be513 HBASE-26064 Introduce a StoreFileTracker to abstract the store file tracking logic
add df57b1c HBASE-25891 Remove dependence on storing WAL filenames for backup (#3359)
add fc5a8fd Revert "HBASE-25891 Remove dependence on storing WAL filenames for backup (#3359)"
add 04c3888 HBASE-25891 Remove dependence on storing WAL filenames for backup (#3359)
add d26bcaa HBASE-25288 Make MasterRpcServices not extends RSRpcServices and also HMaster not extends HRegionServer (#3612)
new 22ac62b HBASE-26064 Introduce a StoreFileTracker to abstract the store file tracking logic
new 488c4cf HBASE-25988 Store the store file list by a file (#3578)
new bd924ab HBASE-26079 Use StoreFileTracker when splitting and merging (#3617)
new 24f48b0 HBASE-26224 Introduce a MigrationStoreFileTracker to support migrating from different store file tracker implementations (#3656)
new d76ba96 HBASE-26246 Persist the StoreFileTracker configurations to TableDescriptor when creating table (#3666)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (9300708)
\
N -- N -- N refs/heads/HBASE-26067 (d76ba96)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/hadoop/hbase/backup/BackupInfo.java | 51 +-
.../hadoop/hbase/backup/impl/BackupManager.java | 23 +-
.../hadoop/hbase/backup/impl/BackupManifest.java | 20 +-
.../hbase/backup/impl/BackupSystemTable.java | 207 +---
.../hbase/backup/impl/FullTableBackupClient.java | 12 +-
.../backup/impl/IncrementalBackupManager.java | 132 +--
.../backup/impl/IncrementalTableBackupClient.java | 8 +-
.../hbase/backup/impl/TableBackupClient.java | 5 +-
.../hbase/backup/master/BackupLogCleaner.java | 98 +-
.../hadoop/hbase/backup/util/BackupUtils.java | 19 +-
.../apache/hadoop/hbase/backup/TestBackupBase.java | 47 +-
.../hadoop/hbase/backup/TestBackupSystemTable.java | 49 +-
.../hbase/backup/master/TestBackupLogCleaner.java | 49 +-
.../src/main/protobuf/Backup.proto | 4 +
.../apache/hadoop/hbase/HBaseRpcServicesBase.java | 386 +++++++
.../org/apache/hadoop/hbase/HBaseServerBase.java | 600 +++++++++++
.../main/java/org/apache/hadoop/hbase/Server.java | 4 +-
.../ipc/AnnotationReadingPriorityFunction.java | 4 +-
.../apache/hadoop/hbase/ipc/NettyRpcServer.java | 3 +-
.../org/apache/hadoop/hbase/master/HMaster.java | 218 ++--
.../hadoop/hbase/master/MasterRpcServices.java | 1134 +++++++++++---------
.../hbase/master/http/MasterDumpServlet.java | 6 -
.../hadoop/hbase/regionserver/HRegionServer.java | 710 ++----------
.../MetricsRegionServerWrapperImpl.java | 12 +-
.../hadoop/hbase/regionserver/RSRpcServices.java | 618 +++--------
.../regionserver/ReplicationSource.java | 6 +-
.../hbase/security/access/AccessChecker.java | 2 +-
.../hbase/util/NettyEventLoopGroupConfig.java | 10 +
.../OOMEChecker.java} | 44 +-
.../org/apache/hadoop/hbase/HBaseTestingUtil.java | 12 -
.../hadoop/hbase/SingleProcessHBaseCluster.java | 12 -
.../hbase/TestMetaUpdatesGoToPriorityQueue.java | 9 +-
.../hbase/client/TestSeparateClientZKCluster.java | 4 +-
.../hadoop/hbase/client/TestTableFavoredNodes.java | 6 -
.../hbase/ipc/TestMasterFifoRpcScheduler.java | 7 +-
.../hadoop/hbase/master/AbstractTestDLS.java | 24 +-
.../hbase/master/TestGetLastFlushedSequenceId.java | 10 +-
.../hbase/master/TestGetReplicationLoad.java | 5 -
.../hadoop/hbase/master/TestMasterMetrics.java | 8 +-
.../hbase/master/TestMasterNotCarryTable.java | 81 --
.../hadoop/hbase/master/TestMasterQosFunction.java | 3 -
.../hadoop/hbase/master/TestRollingRestart.java | 3 -
.../assignment/TestWakeUpUnexpectedProcedure.java | 2 +-
.../hbase/master/http/TestMasterStatusServlet.java | 12 -
.../hbase/namequeues/TestSlowLogAccessor.java | 5 +-
.../hbase/regionserver/TestMultiLogThreshold.java | 1 -
.../hbase/regionserver/TestMutateRowsRecovery.java | 2 +-
.../TestRegionMergeTransactionOnCluster.java | 2 +-
.../hbase/regionserver/TestRegionServerAbort.java | 7 -
.../regionserver/TestRegionServerNoMaster.java | 19 +-
.../regionserver/TestScannerRPCScanMetrics.java | 2 +-
.../TestSplitTransactionOnCluster.java | 2 +-
.../hbase/replication/TestReplicationStatus.java | 2 +-
.../apache/hadoop/hbase/HBaseTestingUtility.java | 15 +-
.../org/apache/hadoop/hbase/MiniHBaseCluster.java | 14 +-
55 files changed, 2230 insertions(+), 2520 deletions(-)
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/HBaseRpcServicesBase.java
create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/HBaseServerBase.java
copy hbase-server/src/main/java/org/apache/hadoop/hbase/{namequeues/DisruptorExceptionHandler.java => util/OOMEChecker.java} (54%)
delete mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterNotCarryTable.java
[hbase] 05/05: HBASE-26246 Persist the StoreFileTracker
configurations to TableDescriptor when creating table (#3666)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-26067
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit d76ba9616c5c092422699efa2fd35a0272b20af8
Author: Wellington Ramos Chevreuil <wc...@apache.org>
AuthorDate: Sun Sep 12 14:14:03 2021 +0100
HBASE-26246 Persist the StoreFileTracker configurations to TableDescriptor when creating table (#3666)
Signed-off-by: Duo Zhang <zh...@apache.org>
---
.../master/procedure/CreateTableProcedure.java | 7 ++++++
.../hbase/regionserver/HRegionFileSystem.java | 2 +-
.../MigrationStoreFileTracker.java | 8 +++++++
.../storefiletracker/StoreFileTracker.java | 8 +++++++
.../storefiletracker/StoreFileTrackerBase.java | 13 +++++++++++
.../storefiletracker/StoreFileTrackerFactory.java | 25 +++++++++++++++-------
.../org/apache/hadoop/hbase/client/TestAdmin.java | 6 ++++++
.../org/apache/hadoop/hbase/client/TestAdmin3.java | 6 ++++++
.../hbase/client/TestAsyncTableAdminApi.java | 6 ++++++
.../hbase/client/TestAsyncTableAdminApi3.java | 6 ++++++
.../procedure/MasterProcedureTestingUtility.java | 7 ++++++
.../master/procedure/TestCreateTableProcedure.java | 17 +++++++++++++++
.../storefiletracker/TestStoreFileTracker.java | 14 +++++++++---
13 files changed, 113 insertions(+), 12 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CreateTableProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CreateTableProcedure.java
index 2313e70..dccea55 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CreateTableProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CreateTableProcedure.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.master.procedure;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -33,10 +34,12 @@ import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionReplicaUtil;
import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.client.TableState;
import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
import org.apache.hadoop.hbase.master.MasterFileSystem;
import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.replication.ReplicationException;
import org.apache.hadoop.hbase.rsgroup.RSGroupInfo;
import org.apache.hadoop.hbase.util.CommonFSUtils;
@@ -285,6 +288,10 @@ public class CreateTableProcedure
getTableName(), (newRegions != null ? newRegions.size() : 0));
}
+ TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableDescriptor);
+ StoreFileTrackerFactory.persistTrackerConfig(env.getMasterConfiguration(), builder);
+ tableDescriptor = builder.build();
+
final MasterCoprocessorHost cpHost = env.getMasterCoprocessorHost();
if (cpHost != null) {
final RegionInfo[] regions = newRegions == null ? null :
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
index 3b053a3..40a42ab 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
@@ -607,7 +607,7 @@ public class HRegionFileSystem {
writeRegionInfoFileContent(conf, fs, regionInfoFile, regionInfoContent);
HRegionFileSystem regionFs = HRegionFileSystem.openRegionFromFileSystem(
env.getMasterConfiguration(), fs, getTableDir(), regionInfo, false);
- insertRegionFilesIntoStoreTracker(allRegionFiles, env, regionFs);
+ insertRegionFilesIntoStoreTracker(allRegionFiles, env, regionFs);
}
return regionDir;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/MigrationStoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/MigrationStoreFileTracker.java
index e486e6d..483a240 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/MigrationStoreFileTracker.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/MigrationStoreFileTracker.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.util.Collection;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.regionserver.StoreContext;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
import org.apache.yetus.audience.InterfaceAudience;
@@ -85,4 +86,11 @@ class MigrationStoreFileTracker extends StoreFileTrackerBase {
throw new UnsupportedOperationException(
"Should not call this method on " + getClass().getSimpleName());
}
+
+ @Override
+ public void persistConfiguration(TableDescriptorBuilder builder) {
+ super.persistConfiguration(builder);
+ builder.setValue(SRC_IMPL, src.getClass().getName());
+ builder.setValue(DST_IMPL, dst.getClass().getName());
+ }
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java
index 0a85abb..81fa1a9 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java
@@ -20,6 +20,8 @@ package org.apache.hadoop.hbase.regionserver.storefiletracker;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
+
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.regionserver.CreateStoreFileWriterParams;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
@@ -71,4 +73,10 @@ public interface StoreFileTracker {
* @return Writer for a new StoreFile
*/
StoreFileWriter createWriter(CreateStoreFileWriterParams params) throws IOException;
+
+ /**
+ * Saves StoreFileTracker implementations specific configs into the table descriptors.
+ * @param builder The table descriptor builder for the given table.
+ */
+ void persistConfiguration(TableDescriptorBuilder builder);
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
index d860f8e..83ebbc7 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
@@ -17,18 +17,22 @@
*/
package org.apache.hadoop.hbase.regionserver.storefiletracker;
+import static org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory.TRACK_IMPL;
+
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.crypto.Encryption;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileContext;
import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
+import org.apache.hadoop.hbase.procedure2.util.StringUtils;
import org.apache.hadoop.hbase.regionserver.CreateStoreFileWriterParams;
import org.apache.hadoop.hbase.regionserver.StoreContext;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
@@ -78,6 +82,15 @@ abstract class StoreFileTrackerBase implements StoreFileTracker {
}
}
+ @Override
+ public void persistConfiguration(TableDescriptorBuilder builder) {
+ if (StringUtils.isEmpty(builder.getValue(TRACK_IMPL))) {
+ String trackerImpl = StoreFileTrackerFactory.
+ getStoreFileTrackerImpl(conf).getName();
+ builder.setValue(TRACK_IMPL, trackerImpl).build();
+ }
+ }
+
private HFileContext createFileContext(Compression.Algorithm compression,
boolean includeMVCCReadpoint, boolean includesTag, Encryption.Context encryptionContext) {
if (compression == null) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
index 2c2b71d..daec74c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
@@ -6,9 +6,7 @@
* 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
- *
+ * 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.
@@ -22,6 +20,7 @@ import org.apache.hadoop.hbase.CompoundConfiguration;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
import org.apache.hadoop.hbase.regionserver.StoreContext;
import org.apache.hadoop.hbase.util.Bytes;
@@ -35,15 +34,17 @@ import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
/**
* Factory method for creating store file tracker.
*/
-@InterfaceAudience.Private
-public final class StoreFileTrackerFactory {
+@InterfaceAudience.Private public final class StoreFileTrackerFactory {
public static final String TRACK_IMPL = "hbase.store.file-tracker.impl";
private static final Logger LOG = LoggerFactory.getLogger(StoreFileTrackerFactory.class);
+ public static Class<? extends StoreFileTracker> getStoreFileTrackerImpl(Configuration conf) {
+ return conf.getClass(TRACK_IMPL, DefaultStoreFileTracker.class, StoreFileTracker.class);
+ }
+
public static StoreFileTracker create(Configuration conf, boolean isPrimaryReplica,
StoreContext ctx) {
- Class<? extends StoreFileTracker> tracker =
- conf.getClass(TRACK_IMPL, DefaultStoreFileTracker.class, StoreFileTracker.class);
+ Class<? extends StoreFileTracker> tracker = getStoreFileTrackerImpl(conf);
LOG.info("instantiating StoreFileTracker impl {}", tracker.getName());
return ReflectionUtils.newInstance(tracker, conf, isPrimaryReplica, ctx);
}
@@ -54,7 +55,7 @@ public final class StoreFileTrackerFactory {
ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family));
StoreContext ctx = StoreContext.getBuilder().withColumnFamilyDescriptor(fDescBuilder.build())
.withRegionFileSystem(regionFs).build();
- return StoreFileTrackerFactory.create(conf, TRACK_IMPL, isPrimaryReplica, ctx);
+ return StoreFileTrackerFactory.create(conf, isPrimaryReplica, ctx);
}
public static Configuration mergeConfigurations(Configuration global, TableDescriptor table,
@@ -76,4 +77,12 @@ public final class StoreFileTrackerFactory {
LOG.info("instantiating StoreFileTracker impl {} as {}", tracker.getName(), configName);
return ReflectionUtils.newInstance(tracker, conf, isPrimaryReplica, ctx);
}
+
+ public static void persistTrackerConfig(Configuration conf, TableDescriptorBuilder builder) {
+ TableDescriptor tableDescriptor = builder.build();
+ ColumnFamilyDescriptor cfDesc = tableDescriptor.getColumnFamilies()[0];
+ StoreContext context = StoreContext.getBuilder().withColumnFamilyDescriptor(cfDesc).build();
+ StoreFileTracker tracker = StoreFileTrackerFactory.create(conf, true, context);
+ tracker.persistConfiguration(builder);
+ }
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin.java
index 2230262..10ab496 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.client;
import static org.apache.hadoop.hbase.HBaseTestingUtil.countRows;
+import static org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory.TRACK_IMPL;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
@@ -39,6 +40,7 @@ import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableExistsException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
@@ -423,6 +425,10 @@ public class TestAdmin extends TestAdminBase {
assertEquals(BLOCK_SIZE, newTableDesc.getColumnFamily(FAMILY_1).getBlocksize());
assertEquals(BLOCK_CACHE, newTableDesc.getColumnFamily(FAMILY_1).isBlockCacheEnabled());
assertEquals(TTL, newTableDesc.getColumnFamily(FAMILY_1).getTimeToLive());
+ // HBASE-26246 introduced persist of store file tracker into table descriptor
+ tableDesc = TableDescriptorBuilder.newBuilder(tableDesc).setValue(TRACK_IMPL,
+ StoreFileTrackerFactory.getStoreFileTrackerImpl(TEST_UTIL.getConfiguration()).getName()).
+ build();
TEST_UTIL.verifyTableDescriptorIgnoreTableName(tableDesc, newTableDesc);
if (preserveSplits) {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin3.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin3.java
index e7be08e..7d40fd1 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin3.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin3.java
@@ -17,6 +17,7 @@
*/
package org.apache.hadoop.hbase.client;
+import static org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory.TRACK_IMPL;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -34,6 +35,7 @@ import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.TableNotDisabledException;
import org.apache.hadoop.hbase.TableNotEnabledException;
import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
@@ -236,6 +238,10 @@ public class TestAdmin3 extends TestAdminBase {
ADMIN.createTable(htd);
Table table = TEST_UTIL.getConnection().getTable(htd.getTableName());
TableDescriptor confirmedHtd = table.getDescriptor();
+ //HBASE-26246 introduced persist of store file tracker into table descriptor
+ htd = TableDescriptorBuilder.newBuilder(htd).setValue(TRACK_IMPL,
+ StoreFileTrackerFactory.getStoreFileTrackerImpl(TEST_UTIL.getConfiguration()).getName()).
+ build();
assertEquals(0, TableDescriptor.COMPARATOR.compare(htd, confirmedHtd));
MetaTableAccessor.fullScanMetaAndPrint(TEST_UTIL.getConnection());
table.close();
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java
index 572a1d5..a1832f4 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.client;
import static org.apache.hadoop.hbase.TableName.META_TABLE_NAME;
+import static org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory.TRACK_IMPL;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -34,6 +35,7 @@ import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.TableExistsException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
@@ -372,6 +374,10 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
assertEquals(BLOCK_SIZE, newTableDesc.getColumnFamily(FAMILY_1).getBlocksize());
assertEquals(BLOCK_CACHE, newTableDesc.getColumnFamily(FAMILY_1).isBlockCacheEnabled());
assertEquals(TTL, newTableDesc.getColumnFamily(FAMILY_1).getTimeToLive());
+ //HBASE-26246 introduced persist of store file tracker into table descriptor
+ tableDesc = TableDescriptorBuilder.newBuilder(tableDesc).setValue(TRACK_IMPL,
+ StoreFileTrackerFactory.getStoreFileTrackerImpl(TEST_UTIL.getConfiguration()).getName()).
+ build();
TEST_UTIL.verifyTableDescriptorIgnoreTableName(tableDesc, newTableDesc);
if (preserveSplits) {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi3.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi3.java
index 4a71baf..cb89bf5 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi3.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi3.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.client;
import static org.apache.hadoop.hbase.TableName.META_TABLE_NAME;
+import static org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory.TRACK_IMPL;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertEquals;
@@ -35,6 +36,7 @@ import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.util.Bytes;
@@ -147,6 +149,10 @@ public class TestAsyncTableAdminApi3 extends TestAsyncAdminBase {
TableDescriptor desc = builder.build();
admin.createTable(desc).join();
TableDescriptor confirmedHtd = admin.getDescriptor(tableName).get();
+ //HBASE-26246 introduced persist of store file tracker into table descriptor
+ desc = TableDescriptorBuilder.newBuilder(desc).setValue(TRACK_IMPL,
+ StoreFileTrackerFactory.getStoreFileTrackerImpl(TEST_UTIL.getConfiguration()).getName()).
+ build();
assertEquals(0, TableDescriptor.COMPARATOR.compare(desc, confirmedHtd));
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureTestingUtility.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureTestingUtility.java
index a320643..2ceec5f 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureTestingUtility.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/MasterProcedureTestingUtility.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.master.procedure;
+import static org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory.TRACK_IMPL;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -60,6 +61,7 @@ import org.apache.hadoop.hbase.procedure2.Procedure;
import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility;
import org.apache.hadoop.hbase.procedure2.StateMachineProcedure;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
@@ -226,6 +228,11 @@ public class MasterProcedureTestingUtility {
assertTrue("family not found " + family[i], htd.getColumnFamily(Bytes.toBytes(family[i])) != null);
}
assertEquals(family.length, htd.getColumnFamilyCount());
+
+ // checks store file tracker impl has been properly set in htd
+ String storeFileTrackerImpl =
+ StoreFileTrackerFactory.getStoreFileTrackerImpl(master.getConfiguration()).getName();
+ assertEquals(storeFileTrackerImpl, htd.getValue(TRACK_IMPL));
}
public static void validateTableDeletion(
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestCreateTableProcedure.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestCreateTableProcedure.java
index 3b57ed5..0bc77f0 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestCreateTableProcedure.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestCreateTableProcedure.java
@@ -17,6 +17,7 @@
*/
package org.apache.hadoop.hbase.master.procedure;
+import static org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory.TRACK_IMPL;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -38,6 +39,7 @@ import org.apache.hadoop.hbase.master.MasterFileSystem;
import org.apache.hadoop.hbase.procedure2.Procedure;
import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.TestStoreFileTracker;
import org.apache.hadoop.hbase.testclassification.MasterTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.Bytes;
@@ -89,6 +91,21 @@ public class TestCreateTableProcedure extends TestTableDDLProcedureBase {
}
@Test
+ public void testCreateWithTrackImpl() throws Exception {
+ final TableName tableName = TableName.valueOf(name.getMethodName());
+ ProcedureExecutor<MasterProcedureEnv> procExec = getMasterProcedureExecutor();
+ TableDescriptor htd = MasterProcedureTestingUtility.createHTD(tableName, F1);
+ String trackerName = TestStoreFileTracker.class.getName();
+ htd = TableDescriptorBuilder.newBuilder(htd).setValue(TRACK_IMPL, trackerName).build();
+ RegionInfo[] regions = ModifyRegionUtils.createRegionInfos(htd, null);
+ long procId = ProcedureTestingUtility.submitAndWait(procExec,
+ new CreateTableProcedure(procExec.getEnvironment(), htd, regions));
+ ProcedureTestingUtility.assertProcNotFailed(procExec.getResult(procId));
+ htd = getMaster().getTableDescriptors().get(tableName);
+ assertEquals(trackerName, htd.getValue(TRACK_IMPL));
+ }
+
+ @Test
public void testCreateWithoutColumnFamily() throws Exception {
final ProcedureExecutor<MasterProcedureEnv> procExec = getMasterProcedureExecutor();
final TableName tableName = TableName.valueOf(name.getMethodName());
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestStoreFileTracker.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestStoreFileTracker.java
index 05ca1fc..1dc9c4e 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestStoreFileTracker.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestStoreFileTracker.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.regionserver.storefiletracker;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -27,6 +28,7 @@ import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.regionserver.StoreContext;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,15 +40,21 @@ public class TestStoreFileTracker extends DefaultStoreFileTracker {
public TestStoreFileTracker(Configuration conf, boolean isPrimaryReplica, StoreContext ctx) {
super(conf, isPrimaryReplica, ctx);
- this.storeId = ctx.getRegionInfo().getEncodedName() + "-" + ctx.getFamily().getNameAsString();
- LOG.info("created storeId: {}", storeId);
- trackedFiles.computeIfAbsent(storeId, v -> new ArrayList<>());
+ if (ctx.getRegionFileSystem() != null) {
+ this.storeId = ctx.getRegionInfo().getEncodedName() + "-" + ctx.getFamily().getNameAsString();
+ LOG.info("created storeId: {}", storeId);
+ trackedFiles.computeIfAbsent(storeId, v -> new ArrayList<>());
+ } else {
+ LOG.info("ctx.getRegionFileSystem() returned null. Leaving storeId null.");
+ }
+
}
@Override
protected void doAddNewStoreFiles(Collection<StoreFileInfo> newFiles) throws IOException {
LOG.info("adding to storeId: {}", storeId);
trackedFiles.get(storeId).addAll(newFiles);
+ trackedFiles.putIfAbsent(storeId, (List<StoreFileInfo>)newFiles);
}
@Override
[hbase] 01/05: HBASE-26064 Introduce a StoreFileTracker to abstract
the store file tracking logic
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-26067
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 22ac62bc41fd584ffb2a56619a00df0e7ac3333c
Author: Duo Zhang <zh...@apache.org>
AuthorDate: Thu Jul 29 18:35:19 2021 +0800
HBASE-26064 Introduce a StoreFileTracker to abstract the store file tracking logic
Signed-off-by: Wellington Chevreuil <wc...@apache.org>
---
.../hadoop/hbase/mob/DefaultMobStoreCompactor.java | 22 +-
.../hadoop/hbase/mob/DefaultMobStoreFlusher.java | 4 +-
.../regionserver/CreateStoreFileWriterParams.java | 134 ++++
.../hbase/regionserver/DateTieredStoreEngine.java | 5 +-
.../hbase/regionserver/DefaultStoreEngine.java | 5 +-
.../hbase/regionserver/DefaultStoreFlusher.java | 11 +-
.../hadoop/hbase/regionserver/HMobStore.java | 3 +-
.../hbase/regionserver/HRegionFileSystem.java | 10 +-
.../apache/hadoop/hbase/regionserver/HStore.java | 699 +++++----------------
.../hadoop/hbase/regionserver/StoreContext.java | 9 +
.../hadoop/hbase/regionserver/StoreEngine.java | 438 ++++++++++++-
.../hbase/regionserver/StoreFileManager.java | 9 +
.../hadoop/hbase/regionserver/StoreFlusher.java | 9 +-
.../hadoop/hbase/regionserver/StoreUtils.java | 36 +-
.../hbase/regionserver/StripeStoreEngine.java | 9 +-
.../hbase/regionserver/StripeStoreFlusher.java | 9 +-
.../compactions/AbstractMultiOutputCompactor.java | 7 +-
.../hbase/regionserver/compactions/Compactor.java | 36 +-
.../regionserver/compactions/DefaultCompactor.java | 16 +-
.../storefiletracker/DefaultStoreFileTracker.java | 61 ++
.../storefiletracker/StoreFileTracker.java | 75 +++
.../storefiletracker/StoreFileTrackerBase.java | 178 ++++++
.../storefiletracker/StoreFileTrackerFactory.java | 35 ++
.../util/compaction/MajorCompactionRequest.java | 1 -
.../org/apache/hadoop/hbase/TestIOFencing.java | 12 +-
.../regionserver/TestCacheOnWriteInSchema.java | 6 +-
.../hbase/regionserver/TestDefaultStoreEngine.java | 5 +-
.../hadoop/hbase/regionserver/TestHRegion.java | 4 +-
.../hadoop/hbase/regionserver/TestHStore.java | 28 +-
.../TestRegionMergeTransactionOnCluster.java | 6 +-
.../regionserver/TestStoreFileRefresherChore.java | 3 +-
.../regionserver/TestStoreScannerClosure.java | 6 +-
.../hbase/regionserver/TestStripeStoreEngine.java | 2 +
.../compactions/TestDateTieredCompactor.java | 12 +-
.../compactions/TestStripeCompactionPolicy.java | 12 +-
.../compactions/TestStripeCompactor.java | 12 +-
36 files changed, 1224 insertions(+), 705 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
index c45fdff..01fe000 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
@@ -29,7 +29,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;
import java.util.Optional;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -144,17 +143,16 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
};
private final CellSinkFactory<StoreFileWriter> writerFactory =
- new CellSinkFactory<StoreFileWriter>() {
- @Override
- public StoreFileWriter createWriter(InternalScanner scanner,
- org.apache.hadoop.hbase.regionserver.compactions.Compactor.FileDetails fd,
- boolean shouldDropBehind, boolean major) throws IOException {
- // make this writer with tags always because of possible new cells with tags.
- return store.createWriterInTmp(fd.maxKeyCount,
- major ? majorCompactionCompression : minorCompactionCompression,
- true, true, true, shouldDropBehind);
- }
- };
+ new CellSinkFactory<StoreFileWriter>() {
+ @Override
+ public StoreFileWriter createWriter(InternalScanner scanner,
+ org.apache.hadoop.hbase.regionserver.compactions.Compactor.FileDetails fd,
+ boolean shouldDropBehind, boolean major) throws IOException {
+ // make this writer with tags always because of possible new cells with tags.
+ return store.getStoreEngine().createWriter(
+ createParams(fd, shouldDropBehind, major).includeMVCCReadpoint(true).includesTag(true));
+ }
+ };
public DefaultMobStoreCompactor(Configuration conf, HStore store) {
super(conf, store);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java
index 480b85c..4a1dc7b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreFlusher.java
@@ -25,7 +25,6 @@ import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
@@ -127,8 +126,7 @@ public class DefaultMobStoreFlusher extends DefaultStoreFlusher {
synchronized (flushLock) {
status.setStatus("Flushing " + store + ": creating writer");
// Write the map out to the disk
- writer = store.createWriterInTmp(cellsCount, store.getColumnFamilyDescriptor().getCompressionType(),
- false, true, true, false);
+ writer = createWriter(snapshot, true);
IOException e = null;
try {
// It's a mob store, flush the cells in a mob way. This is the difference of flushing
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CreateStoreFileWriterParams.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CreateStoreFileWriterParams.java
new file mode 100644
index 0000000..10cd9f0
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CreateStoreFileWriterParams.java
@@ -0,0 +1,134 @@
+/**
+ * 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.hadoop.hbase.regionserver;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public final class CreateStoreFileWriterParams {
+
+ private long maxKeyCount;
+
+ private Compression.Algorithm compression;
+
+ private boolean isCompaction;
+
+ private boolean includeMVCCReadpoint;
+
+ private boolean includesTag;
+
+ private boolean shouldDropBehind;
+
+ private long totalCompactedFilesSize = -1;
+
+ private String fileStoragePolicy = HConstants.EMPTY_STRING;
+
+ private CreateStoreFileWriterParams() {
+ }
+
+ public long maxKeyCount() {
+ return maxKeyCount;
+ }
+
+ public CreateStoreFileWriterParams maxKeyCount(long maxKeyCount) {
+ this.maxKeyCount = maxKeyCount;
+ return this;
+ }
+
+ public Compression.Algorithm compression() {
+ return compression;
+ }
+
+ /**
+ * Set the compression algorithm to use
+ */
+ public CreateStoreFileWriterParams compression(Compression.Algorithm compression) {
+ this.compression = compression;
+ return this;
+ }
+
+ public boolean isCompaction() {
+ return isCompaction;
+ }
+
+ /**
+ * Whether we are creating a new file in a compaction
+ */
+ public CreateStoreFileWriterParams isCompaction(boolean isCompaction) {
+ this.isCompaction = isCompaction;
+ return this;
+ }
+
+ public boolean includeMVCCReadpoint() {
+ return includeMVCCReadpoint;
+ }
+
+ /**
+ * Whether to include MVCC or not
+ */
+ public CreateStoreFileWriterParams includeMVCCReadpoint(boolean includeMVCCReadpoint) {
+ this.includeMVCCReadpoint = includeMVCCReadpoint;
+ return this;
+ }
+
+ public boolean includesTag() {
+ return includesTag;
+ }
+
+ /**
+ * Whether to includesTag or not
+ */
+ public CreateStoreFileWriterParams includesTag(boolean includesTag) {
+ this.includesTag = includesTag;
+ return this;
+ }
+
+ public boolean shouldDropBehind() {
+ return shouldDropBehind;
+ }
+
+ public CreateStoreFileWriterParams shouldDropBehind(boolean shouldDropBehind) {
+ this.shouldDropBehind = shouldDropBehind;
+ return this;
+ }
+
+ public long totalCompactedFilesSize() {
+ return totalCompactedFilesSize;
+ }
+
+ public CreateStoreFileWriterParams totalCompactedFilesSize(long totalCompactedFilesSize) {
+ this.totalCompactedFilesSize = totalCompactedFilesSize;
+ return this;
+ }
+
+ public String fileStoragePolicy() {
+ return fileStoragePolicy;
+ }
+
+ public CreateStoreFileWriterParams fileStoragePolicy(String fileStoragePolicy) {
+ this.fileStoragePolicy = fileStoragePolicy;
+ return this;
+ }
+
+ public static CreateStoreFileWriterParams create() {
+ return new CreateStoreFileWriterParams();
+ }
+
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DateTieredStoreEngine.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DateTieredStoreEngine.java
index 1df953d..7422d91 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DateTieredStoreEngine.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DateTieredStoreEngine.java
@@ -19,18 +19,17 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.List;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CellComparator;
-import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
-import org.apache.yetus.audience.InterfaceAudience;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
import org.apache.hadoop.hbase.regionserver.compactions.DateTieredCompactionPolicy;
import org.apache.hadoop.hbase.regionserver.compactions.DateTieredCompactionRequest;
import org.apache.hadoop.hbase.regionserver.compactions.DateTieredCompactor;
import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
import org.apache.hadoop.hbase.security.User;
+import org.apache.yetus.audience.InterfaceAudience;
/**
* HBASE-15400 This store engine allows us to store data in date tiered layout with exponential
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreEngine.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreEngine.java
index 58f8bbb..693b9c9 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreEngine.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreEngine.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.List;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CellComparator;
@@ -39,8 +38,8 @@ import org.apache.yetus.audience.InterfaceAudience;
* their derivatives.
*/
@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
-public class DefaultStoreEngine extends StoreEngine<
- DefaultStoreFlusher, RatioBasedCompactionPolicy, DefaultCompactor, DefaultStoreFileManager> {
+public class DefaultStoreEngine extends StoreEngine<DefaultStoreFlusher,
+ RatioBasedCompactionPolicy, DefaultCompactor, DefaultStoreFileManager> {
public static final String DEFAULT_STORE_FLUSHER_CLASS_KEY =
"hbase.hstore.defaultengine.storeflusher.class";
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreFlusher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreFlusher.java
index a7d7fb1..306760d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreFlusher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultStoreFlusher.java
@@ -21,15 +21,14 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-
-import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
import org.apache.hadoop.util.StringUtils;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Default implementation of StoreFlusher.
@@ -60,9 +59,7 @@ public class DefaultStoreFlusher extends StoreFlusher {
synchronized (flushLock) {
status.setStatus("Flushing " + store + ": creating writer");
// Write the map out to the disk
- writer = store.createWriterInTmp(cellsCount,
- store.getColumnFamilyDescriptor().getCompressionType(), false, true,
- snapshot.isTagsPresent(), false);
+ writer = createWriter(snapshot, false);
IOException e = null;
try {
performFlush(scanner, writer, throughputController);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HMobStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HMobStore.java
index 7ce7f03..b00a50c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HMobStore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HMobStore.java
@@ -28,7 +28,6 @@ import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -158,7 +157,7 @@ public class HMobStore extends HStore {
protected StoreEngine<?, ?, ?, ?> createStoreEngine(HStore store, Configuration conf,
CellComparator cellComparator) throws IOException {
MobStoreEngine engine = new MobStoreEngine();
- engine.createComponents(conf, store, cellComparator);
+ engine.createComponentsOnce(conf, store, cellComparator);
return engine;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
index 5b26f9a..6ab9b0f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
@@ -142,7 +142,7 @@ public class HRegionFileSystem {
// Temp Helpers
// ===========================================================================
/** @return {@link Path} to the region's temp directory, used for file creations */
- Path getTempDir() {
+ public Path getTempDir() {
return new Path(getRegionDir(), REGION_TEMP_DIR);
}
@@ -237,11 +237,7 @@ public class HRegionFileSystem {
* @param familyName Column Family Name
* @return a set of {@link StoreFileInfo} for the specified family.
*/
- public Collection<StoreFileInfo> getStoreFiles(final byte[] familyName) throws IOException {
- return getStoreFiles(Bytes.toString(familyName));
- }
-
- public Collection<StoreFileInfo> getStoreFiles(final String familyName) throws IOException {
+ public List<StoreFileInfo> getStoreFiles(final String familyName) throws IOException {
return getStoreFiles(familyName, true);
}
@@ -251,7 +247,7 @@ public class HRegionFileSystem {
* @param familyName Column Family Name
* @return a set of {@link StoreFileInfo} for the specified family.
*/
- public Collection<StoreFileInfo> getStoreFiles(final String familyName, final boolean validate)
+ public List<StoreFileInfo> getStoreFiles(final String familyName, final boolean validate)
throws IOException {
Path familyDir = getStoreDir(familyName);
FileStatus[] files = CommonFSUtils.listStatus(this.fs, familyDir);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
index c4e34be..34365b9 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
@@ -17,6 +17,7 @@
*/
package org.apache.hadoop.hbase.regionserver;
+import com.google.errorprone.annotations.RestrictedApi;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.InetSocketAddress;
@@ -47,8 +48,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.function.Predicate;
import java.util.function.ToLongFunction;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
@@ -70,17 +69,12 @@ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.conf.ConfigurationManager;
import org.apache.hadoop.hbase.conf.PropagatingConfigurationObserver;
import org.apache.hadoop.hbase.io.HeapSize;
-import org.apache.hadoop.hbase.io.compress.Compression;
-import org.apache.hadoop.hbase.io.crypto.Encryption;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.HFile;
-import org.apache.hadoop.hbase.io.hfile.HFileContext;
-import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoder;
import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoderImpl;
import org.apache.hadoop.hbase.io.hfile.HFileScanner;
import org.apache.hadoop.hbase.io.hfile.InvalidHFileException;
-import org.apache.hadoop.hbase.log.HBaseMarkers;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.quotas.RegionSizeStore;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
@@ -110,7 +104,6 @@ import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableCollection
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableList;
import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
-import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;
import org.apache.hbase.thirdparty.org.apache.commons.collections4.IterableUtils;
@@ -166,16 +159,6 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
private boolean cacheOnWriteLogged;
/**
- * RWLock for store operations.
- * Locked in shared mode when the list of component stores is looked at:
- * - all reads/writes to table data
- * - checking for split
- * Locked in exclusive mode when the list of component stores is modified:
- * - closing
- * - completing a compaction
- */
- final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
- /**
* Lock specific to archiving compacted store files. This avoids races around
* the combination of retrieving the list of compacted files and moving them to
* the archive directory. Since this is usually a background process (other than
@@ -290,14 +273,8 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
this.storeEngine = createStoreEngine(this, this.conf, region.getCellComparator());
- List<HStoreFile> hStoreFiles = loadStoreFiles(warmup);
- // Move the storeSize calculation out of loadStoreFiles() method, because the secondary read
- // replica's refreshStoreFiles() will also use loadStoreFiles() to refresh its store files and
- // update the storeSize in the refreshStoreSizeAndTotalBytes() finally (just like compaction) , so
- // no need calculate the storeSize twice.
- this.storeSize.addAndGet(getStorefilesSize(hStoreFiles, sf -> true));
- this.totalUncompressedBytes.addAndGet(getTotalUncompressedBytes(hStoreFiles));
- this.storeEngine.getStoreFileManager().loadFiles(hStoreFiles);
+ storeEngine.initialize(warmup);
+ refreshStoreSizeAndTotalBytes();
flushRetriesNumber = conf.getInt(
"hbase.hstore.flush.retries.number", DEFAULT_FLUSH_RETRIES_NUMBER);
@@ -513,102 +490,18 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
this.dataBlockEncoder = blockEncoder;
}
- /**
- * Creates an unsorted list of StoreFile loaded in parallel
- * from the given directory.
- */
- private List<HStoreFile> loadStoreFiles(boolean warmup) throws IOException {
- Collection<StoreFileInfo> files = getRegionFileSystem().getStoreFiles(getColumnFamilyName());
- return openStoreFiles(files, warmup);
- }
-
- private List<HStoreFile> openStoreFiles(Collection<StoreFileInfo> files, boolean warmup)
- throws IOException {
- if (CollectionUtils.isEmpty(files)) {
- return Collections.emptyList();
- }
- // initialize the thread pool for opening store files in parallel..
- ThreadPoolExecutor storeFileOpenerThreadPool =
- this.region.getStoreFileOpenAndCloseThreadPool("StoreFileOpener-"
- + this.region.getRegionInfo().getEncodedName() + "-" + this.getColumnFamilyName());
- CompletionService<HStoreFile> completionService =
- new ExecutorCompletionService<>(storeFileOpenerThreadPool);
-
- int totalValidStoreFile = 0;
- for (StoreFileInfo storeFileInfo : files) {
- // open each store file in parallel
- completionService.submit(() -> this.createStoreFileAndReader(storeFileInfo));
- totalValidStoreFile++;
- }
-
- Set<String> compactedStoreFiles = new HashSet<>();
- ArrayList<HStoreFile> results = new ArrayList<>(files.size());
- IOException ioe = null;
- try {
- for (int i = 0; i < totalValidStoreFile; i++) {
- try {
- HStoreFile storeFile = completionService.take().get();
- if (storeFile != null) {
- LOG.debug("loaded {}", storeFile);
- results.add(storeFile);
- compactedStoreFiles.addAll(storeFile.getCompactedStoreFiles());
- }
- } catch (InterruptedException e) {
- if (ioe == null) {
- ioe = new InterruptedIOException(e.getMessage());
- }
- } catch (ExecutionException e) {
- if (ioe == null) {
- ioe = new IOException(e.getCause());
- }
- }
- }
- } finally {
- storeFileOpenerThreadPool.shutdownNow();
- }
- if (ioe != null) {
- // close StoreFile readers
- boolean evictOnClose =
- getCacheConfig() != null? getCacheConfig().shouldEvictOnClose(): true;
- for (HStoreFile file : results) {
- try {
- if (file != null) {
- file.closeStoreFile(evictOnClose);
- }
- } catch (IOException e) {
- LOG.warn("Could not close store file {}", file, e);
- }
- }
- throw ioe;
- }
-
- // Should not archive the compacted store files when region warmup. See HBASE-22163.
- if (!warmup) {
- // Remove the compacted files from result
- List<HStoreFile> filesToRemove = new ArrayList<>(compactedStoreFiles.size());
- for (HStoreFile storeFile : results) {
- if (compactedStoreFiles.contains(storeFile.getPath().getName())) {
- LOG.warn("Clearing the compacted storefile {} from {}", storeFile, this);
- storeFile.getReader().close(storeFile.getCacheConf() != null ?
- storeFile.getCacheConf().shouldEvictOnClose() : true);
- filesToRemove.add(storeFile);
- }
- }
- results.removeAll(filesToRemove);
- if (!filesToRemove.isEmpty() && this.isPrimaryReplicaStore()) {
- LOG.debug("Moving the files {} to archive", filesToRemove);
- getRegionFileSystem().removeStoreFiles(this.getColumnFamilyDescriptor().getNameAsString(),
- filesToRemove);
- }
- }
-
- return results;
+ private void postRefreshStoreFiles() throws IOException {
+ // Advance the memstore read point to be at least the new store files seqIds so that
+ // readers might pick it up. This assumes that the store is not getting any writes (otherwise
+ // in-flight transactions might be made visible)
+ getMaxSequenceId().ifPresent(region.getMVCC()::advanceTo);
+ refreshStoreSizeAndTotalBytes();
}
@Override
public void refreshStoreFiles() throws IOException {
- Collection<StoreFileInfo> newFiles = getRegionFileSystem().getStoreFiles(getColumnFamilyName());
- refreshStoreFilesInternal(newFiles);
+ storeEngine.refreshStoreFiles();
+ postRefreshStoreFiles();
}
/**
@@ -616,89 +509,8 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
* region replicas to keep up to date with the primary region files.
*/
public void refreshStoreFiles(Collection<String> newFiles) throws IOException {
- List<StoreFileInfo> storeFiles = new ArrayList<>(newFiles.size());
- for (String file : newFiles) {
- storeFiles.add(getRegionFileSystem().getStoreFileInfo(getColumnFamilyName(), file));
- }
- refreshStoreFilesInternal(storeFiles);
- }
-
- /**
- * Checks the underlying store files, and opens the files that have not
- * been opened, and removes the store file readers for store files no longer
- * available. Mainly used by secondary region replicas to keep up to date with
- * the primary region files.
- */
- private void refreshStoreFilesInternal(Collection<StoreFileInfo> newFiles) throws IOException {
- StoreFileManager sfm = storeEngine.getStoreFileManager();
- Collection<HStoreFile> currentFiles = sfm.getStorefiles();
- Collection<HStoreFile> compactedFiles = sfm.getCompactedfiles();
- if (currentFiles == null) {
- currentFiles = Collections.emptySet();
- }
- if (newFiles == null) {
- newFiles = Collections.emptySet();
- }
- if (compactedFiles == null) {
- compactedFiles = Collections.emptySet();
- }
-
- HashMap<StoreFileInfo, HStoreFile> currentFilesSet = new HashMap<>(currentFiles.size());
- for (HStoreFile sf : currentFiles) {
- currentFilesSet.put(sf.getFileInfo(), sf);
- }
- HashMap<StoreFileInfo, HStoreFile> compactedFilesSet = new HashMap<>(compactedFiles.size());
- for (HStoreFile sf : compactedFiles) {
- compactedFilesSet.put(sf.getFileInfo(), sf);
- }
-
- Set<StoreFileInfo> newFilesSet = new HashSet<StoreFileInfo>(newFiles);
- // Exclude the files that have already been compacted
- newFilesSet = Sets.difference(newFilesSet, compactedFilesSet.keySet());
- Set<StoreFileInfo> toBeAddedFiles = Sets.difference(newFilesSet, currentFilesSet.keySet());
- Set<StoreFileInfo> toBeRemovedFiles = Sets.difference(currentFilesSet.keySet(), newFilesSet);
-
- if (toBeAddedFiles.isEmpty() && toBeRemovedFiles.isEmpty()) {
- return;
- }
-
- LOG.info("Refreshing store files for " + this + " files to add: "
- + toBeAddedFiles + " files to remove: " + toBeRemovedFiles);
-
- Set<HStoreFile> toBeRemovedStoreFiles = new HashSet<>(toBeRemovedFiles.size());
- for (StoreFileInfo sfi : toBeRemovedFiles) {
- toBeRemovedStoreFiles.add(currentFilesSet.get(sfi));
- }
-
- // try to open the files
- List<HStoreFile> openedFiles = openStoreFiles(toBeAddedFiles, false);
-
- // propogate the file changes to the underlying store file manager
- replaceStoreFiles(toBeRemovedStoreFiles, openedFiles); //won't throw an exception
-
- // Advance the memstore read point to be at least the new store files seqIds so that
- // readers might pick it up. This assumes that the store is not getting any writes (otherwise
- // in-flight transactions might be made visible)
- if (!toBeAddedFiles.isEmpty()) {
- // we must have the max sequence id here as we do have several store files
- region.getMVCC().advanceTo(this.getMaxSequenceId().getAsLong());
- }
-
- refreshStoreSizeAndTotalBytes();
- }
-
- protected HStoreFile createStoreFileAndReader(final Path p) throws IOException {
- StoreFileInfo info = new StoreFileInfo(conf, this.getFileSystem(),
- p, isPrimaryReplicaStore());
- return createStoreFileAndReader(info);
- }
-
- private HStoreFile createStoreFileAndReader(StoreFileInfo info) throws IOException {
- info.setRegionCoprocessorHost(this.region.getCoprocessorHost());
- HStoreFile storeFile = new HStoreFile(info, getColumnFamilyDescriptor().getBloomFilterType(),
- getCacheConfig());
- storeFile.initReader();
- return storeFile;
+ storeEngine.refreshStoreFiles(newFiles);
+ postRefreshStoreFiles();
}
/**
@@ -721,7 +533,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
* Adds a value to the memstore
*/
public void add(final Cell cell, MemStoreSizing memstoreSizing) {
- lock.readLock().lock();
+ storeEngine.readLock();
try {
if (this.currentParallelPutCount.getAndIncrement() > this.parallelPutCountPrintThreshold) {
LOG.trace("tableName={}, encodedName={}, columnFamilyName={} is too busy!",
@@ -729,7 +541,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
this.memstore.add(cell, memstoreSizing);
} finally {
- lock.readLock().unlock();
+ storeEngine.readUnlock();
currentParallelPutCount.decrementAndGet();
}
}
@@ -738,7 +550,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
* Adds the specified value to the memstore
*/
public void add(final Iterable<Cell> cells, MemStoreSizing memstoreSizing) {
- lock.readLock().lock();
+ storeEngine.readLock();
try {
if (this.currentParallelPutCount.getAndIncrement() > this.parallelPutCountPrintThreshold) {
LOG.trace("tableName={}, encodedName={}, columnFamilyName={} is too busy!",
@@ -746,7 +558,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
memstore.add(cells, memstoreSizing);
} finally {
- lock.readLock().unlock();
+ storeEngine.readUnlock();
currentParallelPutCount.decrementAndGet();
}
}
@@ -869,17 +681,16 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
LOG.info("Loaded HFile " + srcPath + " into " + this + " as "
+ dstPath + " - updating store file list.");
- HStoreFile sf = createStoreFileAndReader(dstPath);
+ HStoreFile sf = storeEngine.createStoreFileAndReader(dstPath);
bulkLoadHFile(sf);
- LOG.info("Successfully loaded {} into {} (new location: {})",
- srcPath, this, dstPath);
+ LOG.info("Successfully loaded {} into {} (new location: {})", srcPath, this, dstPath);
return dstPath;
}
public void bulkLoadHFile(StoreFileInfo fileInfo) throws IOException {
- HStoreFile sf = createStoreFileAndReader(fileInfo);
+ HStoreFile sf = storeEngine.createStoreFileAndReader(fileInfo);
bulkLoadHFile(sf);
}
@@ -887,28 +698,74 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
StoreFileReader r = sf.getReader();
this.storeSize.addAndGet(r.length());
this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
-
- // Append the new storefile into the list
- this.lock.writeLock().lock();
- try {
- this.storeEngine.getStoreFileManager().insertNewFiles(Lists.newArrayList(sf));
- } finally {
- // We need the lock, as long as we are updating the storeFiles
- // or changing the memstore. Let us release it before calling
- // notifyChangeReadersObservers. See HBASE-4485 for a possible
- // deadlock scenario that could have happened if continue to hold
- // the lock.
- this.lock.writeLock().unlock();
- }
+ storeEngine.addStoreFiles(Lists.newArrayList(sf));
LOG.info("Loaded HFile " + sf.getFileInfo() + " into " + this);
if (LOG.isTraceEnabled()) {
- String traceMessage = "BULK LOAD time,size,store size,store files ["
- + EnvironmentEdgeManager.currentTime() + "," + r.length() + "," + storeSize
- + "," + storeEngine.getStoreFileManager().getStorefileCount() + "]";
+ String traceMessage = "BULK LOAD time,size,store size,store files [" +
+ EnvironmentEdgeManager.currentTime() + "," + r.length() + "," + storeSize + "," +
+ storeEngine.getStoreFileManager().getStorefileCount() + "]";
LOG.trace(traceMessage);
}
}
+ private ImmutableCollection<HStoreFile> closeWithoutLock() throws IOException {
+ // Clear so metrics doesn't find them.
+ ImmutableCollection<HStoreFile> result = storeEngine.getStoreFileManager().clearFiles();
+ Collection<HStoreFile> compactedfiles = storeEngine.getStoreFileManager().clearCompactedFiles();
+ // clear the compacted files
+ if (CollectionUtils.isNotEmpty(compactedfiles)) {
+ removeCompactedfiles(compactedfiles,
+ getCacheConfig() != null ? getCacheConfig().shouldEvictOnClose() : true);
+ }
+ if (!result.isEmpty()) {
+ // initialize the thread pool for closing store files in parallel.
+ ThreadPoolExecutor storeFileCloserThreadPool =
+ this.region.getStoreFileOpenAndCloseThreadPool("StoreFileCloser-" +
+ this.region.getRegionInfo().getEncodedName() + "-" + this.getColumnFamilyName());
+
+ // close each store file in parallel
+ CompletionService<Void> completionService =
+ new ExecutorCompletionService<>(storeFileCloserThreadPool);
+ for (HStoreFile f : result) {
+ completionService.submit(new Callable<Void>() {
+ @Override
+ public Void call() throws IOException {
+ boolean evictOnClose =
+ getCacheConfig() != null ? getCacheConfig().shouldEvictOnClose() : true;
+ f.closeStoreFile(evictOnClose);
+ return null;
+ }
+ });
+ }
+
+ IOException ioe = null;
+ try {
+ for (int i = 0; i < result.size(); i++) {
+ try {
+ Future<Void> future = completionService.take();
+ future.get();
+ } catch (InterruptedException e) {
+ if (ioe == null) {
+ ioe = new InterruptedIOException();
+ ioe.initCause(e);
+ }
+ } catch (ExecutionException e) {
+ if (ioe == null) {
+ ioe = new IOException(e.getCause());
+ }
+ }
+ }
+ } finally {
+ storeFileCloserThreadPool.shutdownNow();
+ }
+ if (ioe != null) {
+ throw ioe;
+ }
+ }
+ LOG.trace("Closed {}", this);
+ return result;
+ }
+
/**
* Close all the readers We don't need to worry about subsequent requests because the Region holds
* a write lock that will prevent any more reads or writes.
@@ -916,67 +773,18 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
* @throws IOException on failure
*/
public ImmutableCollection<HStoreFile> close() throws IOException {
+ // findbugs can not recognize storeEngine.writeLock is just a lock operation so it will report
+ // UL_UNRELEASED_LOCK_EXCEPTION_PATH, so here we have to use two try finally...
+ // Change later if findbugs becomes smarter in the future.
this.archiveLock.lock();
- this.lock.writeLock().lock();
try {
- // Clear so metrics doesn't find them.
- ImmutableCollection<HStoreFile> result = storeEngine.getStoreFileManager().clearFiles();
- Collection<HStoreFile> compactedfiles =
- storeEngine.getStoreFileManager().clearCompactedFiles();
- // clear the compacted files
- if (CollectionUtils.isNotEmpty(compactedfiles)) {
- removeCompactedfiles(compactedfiles, getCacheConfig() != null ?
- getCacheConfig().shouldEvictOnClose() : true);
- }
- if (!result.isEmpty()) {
- // initialize the thread pool for closing store files in parallel.
- ThreadPoolExecutor storeFileCloserThreadPool = this.region
- .getStoreFileOpenAndCloseThreadPool("StoreFileCloser-"
- + this.region.getRegionInfo().getEncodedName() + "-" + this.getColumnFamilyName());
-
- // close each store file in parallel
- CompletionService<Void> completionService =
- new ExecutorCompletionService<>(storeFileCloserThreadPool);
- for (HStoreFile f : result) {
- completionService.submit(new Callable<Void>() {
- @Override
- public Void call() throws IOException {
- boolean evictOnClose =
- getCacheConfig() != null? getCacheConfig().shouldEvictOnClose(): true;
- f.closeStoreFile(evictOnClose);
- return null;
- }
- });
- }
-
- IOException ioe = null;
- try {
- for (int i = 0; i < result.size(); i++) {
- try {
- Future<Void> future = completionService.take();
- future.get();
- } catch (InterruptedException e) {
- if (ioe == null) {
- ioe = new InterruptedIOException();
- ioe.initCause(e);
- }
- } catch (ExecutionException e) {
- if (ioe == null) {
- ioe = new IOException(e.getCause());
- }
- }
- }
- } finally {
- storeFileCloserThreadPool.shutdownNow();
- }
- if (ioe != null) {
- throw ioe;
- }
+ this.storeEngine.writeLock();
+ try {
+ return closeWithoutLock();
+ } finally {
+ this.storeEngine.writeUnlock();
}
- LOG.trace("Closed {}", this);
- return result;
} finally {
- this.lock.writeLock().unlock();
this.archiveLock.unlock();
}
}
@@ -1006,7 +814,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
try {
for (Path pathName : pathNames) {
lastPathName = pathName;
- validateStoreFile(pathName);
+ storeEngine.validateStoreFile(pathName);
}
return pathNames;
} catch (Exception e) {
@@ -1052,167 +860,18 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
Path dstPath = getRegionFileSystem().commitStoreFile(getColumnFamilyName(), path);
- HStoreFile sf = createStoreFileAndReader(dstPath);
+ HStoreFile sf = storeEngine.createStoreFileAndReader(dstPath);
StoreFileReader r = sf.getReader();
this.storeSize.addAndGet(r.length());
this.totalUncompressedBytes.addAndGet(r.getTotalUncompressedBytes());
- this.lock.writeLock().lock();
- try {
- this.storeEngine.getStoreFileManager().insertNewFiles(Lists.newArrayList(sf));
- } finally {
- this.lock.writeLock().unlock();
- }
+ storeEngine.addStoreFiles(Lists.newArrayList(sf));
LOG.info("Loaded recovered hfile to {}, entries={}, sequenceid={}, filesize={}", sf,
r.getEntries(), r.getSequenceID(), TraditionalBinaryPrefix.long2String(r.length(), "B", 1));
return sf;
}
- /**
- * Commit the given {@code files}.
- * <p/>
- * We will move the file into data directory, and open it.
- * @param files the files want to commit
- * @param validate whether to validate the store files
- * @return the committed store files
- */
- private List<HStoreFile> commitStoreFiles(List<Path> files, boolean validate) throws IOException {
- List<HStoreFile> committedFiles = new ArrayList<>(files.size());
- HRegionFileSystem hfs = getRegionFileSystem();
- String familyName = getColumnFamilyName();
- for (Path file : files) {
- try {
- if (validate) {
- validateStoreFile(file);
- }
- Path committedPath = hfs.commitStoreFile(familyName, file);
- HStoreFile sf = createStoreFileAndReader(committedPath);
- committedFiles.add(sf);
- } catch (IOException e) {
- LOG.error("Failed to commit store file {}", file, e);
- // Try to delete the files we have committed before.
- // It is OK to fail when deleting as leaving the file there does not cause any data
- // corruption problem. It just introduces some duplicated data which may impact read
- // performance a little when reading before compaction.
- for (HStoreFile sf : committedFiles) {
- Path pathToDelete = sf.getPath();
- try {
- sf.deleteStoreFile();
- } catch (IOException deleteEx) {
- LOG.warn(HBaseMarkers.FATAL, "Failed to delete committed store file {}", pathToDelete,
- deleteEx);
- }
- }
- throw new IOException("Failed to commit the flush", e);
- }
- }
- return committedFiles;
- }
-
- public StoreFileWriter createWriterInTmp(long maxKeyCount, Compression.Algorithm compression,
- boolean isCompaction, boolean includeMVCCReadpoint, boolean includesTag,
- boolean shouldDropBehind) throws IOException {
- return createWriterInTmp(maxKeyCount, compression, isCompaction, includeMVCCReadpoint,
- includesTag, shouldDropBehind, -1, HConstants.EMPTY_STRING);
- }
-
- /**
- * @param compression Compression algorithm to use
- * @param isCompaction whether we are creating a new file in a compaction
- * @param includeMVCCReadpoint - whether to include MVCC or not
- * @param includesTag - includesTag or not
- * @return Writer for a new StoreFile in the tmp dir.
- */
- // TODO : allow the Writer factory to create Writers of ShipperListener type only in case of
- // compaction
- public StoreFileWriter createWriterInTmp(long maxKeyCount, Compression.Algorithm compression,
- boolean isCompaction, boolean includeMVCCReadpoint, boolean includesTag,
- boolean shouldDropBehind, long totalCompactedFilesSize, String fileStoragePolicy)
- throws IOException {
- // creating new cache config for each new writer
- final CacheConfig cacheConf = getCacheConfig();
- final CacheConfig writerCacheConf = new CacheConfig(cacheConf);
- if (isCompaction) {
- // Don't cache data on write on compactions, unless specifically configured to do so
- // Cache only when total file size remains lower than configured threshold
- final boolean cacheCompactedBlocksOnWrite =
- getCacheConfig().shouldCacheCompactedBlocksOnWrite();
- // if data blocks are to be cached on write
- // during compaction, we should forcefully
- // cache index and bloom blocks as well
- if (cacheCompactedBlocksOnWrite && totalCompactedFilesSize <= cacheConf
- .getCacheCompactedBlocksOnWriteThreshold()) {
- writerCacheConf.enableCacheOnWrite();
- if (!cacheOnWriteLogged) {
- LOG.info("For {} , cacheCompactedBlocksOnWrite is true, hence enabled " +
- "cacheOnWrite for Data blocks, Index blocks and Bloom filter blocks", this);
- cacheOnWriteLogged = true;
- }
- } else {
- writerCacheConf.setCacheDataOnWrite(false);
- if (totalCompactedFilesSize > cacheConf.getCacheCompactedBlocksOnWriteThreshold()) {
- // checking condition once again for logging
- LOG.debug(
- "For {}, setting cacheCompactedBlocksOnWrite as false as total size of compacted "
- + "files - {}, is greater than cacheCompactedBlocksOnWriteThreshold - {}",
- this, totalCompactedFilesSize,
- cacheConf.getCacheCompactedBlocksOnWriteThreshold());
- }
- }
- } else {
- final boolean shouldCacheDataOnWrite = cacheConf.shouldCacheDataOnWrite();
- if (shouldCacheDataOnWrite) {
- writerCacheConf.enableCacheOnWrite();
- if (!cacheOnWriteLogged) {
- LOG.info("For {} , cacheDataOnWrite is true, hence enabled cacheOnWrite for " +
- "Index blocks and Bloom filter blocks", this);
- cacheOnWriteLogged = true;
- }
- }
- }
- Encryption.Context encryptionContext = storeContext.getEncryptionContext();
- HFileContext hFileContext = createFileContext(compression, includeMVCCReadpoint, includesTag,
- encryptionContext);
- Path familyTempDir = new Path(getRegionFileSystem().getTempDir(), getColumnFamilyName());
- StoreFileWriter.Builder builder =
- new StoreFileWriter.Builder(conf, writerCacheConf, getFileSystem())
- .withOutputDir(familyTempDir)
- .withBloomType(storeContext.getBloomFilterType())
- .withMaxKeyCount(maxKeyCount)
- .withFavoredNodes(storeContext.getFavoredNodes())
- .withFileContext(hFileContext)
- .withShouldDropCacheBehind(shouldDropBehind)
- .withCompactedFilesSupplier(storeContext.getCompactedFilesSupplier())
- .withFileStoragePolicy(fileStoragePolicy);
- return builder.build();
- }
-
- private HFileContext createFileContext(Compression.Algorithm compression,
- boolean includeMVCCReadpoint, boolean includesTag, Encryption.Context encryptionContext) {
- if (compression == null) {
- compression = HFile.DEFAULT_COMPRESSION_ALGORITHM;
- }
- ColumnFamilyDescriptor family = getColumnFamilyDescriptor();
- HFileContext hFileContext = new HFileContextBuilder()
- .withIncludesMvcc(includeMVCCReadpoint)
- .withIncludesTags(includesTag)
- .withCompression(compression)
- .withCompressTags(family.isCompressTags())
- .withChecksumType(StoreUtils.getChecksumType(conf))
- .withBytesPerCheckSum(StoreUtils.getBytesPerChecksum(conf))
- .withBlockSize(family.getBlocksize())
- .withHBaseCheckSum(true)
- .withDataBlockEncoding(family.getDataBlockEncoding())
- .withEncryptionContext(encryptionContext)
- .withCreateTime(EnvironmentEdgeManager.currentTime())
- .withColumnFamily(getColumnFamilyDescriptor().getName())
- .withTableName(getTableName().getName())
- .withCellComparator(getComparator())
- .build();
- return hFileContext;
- }
-
private long getTotalSize(Collection<HStoreFile> sfs) {
return sfs.stream().mapToLong(sf -> sf.getReader().length()).sum();
}
@@ -1222,18 +881,8 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
* @param sfs Store files
* @return Whether compaction is required.
*/
- private boolean updateStorefiles(List<HStoreFile> sfs, long snapshotId) throws IOException {
- this.lock.writeLock().lock();
- try {
- this.storeEngine.getStoreFileManager().insertNewFiles(sfs);
- } finally {
- // We need the lock, as long as we are updating the storeFiles
- // or changing the memstore. Let us release it before calling
- // notifyChangeReadersObservers. See HBASE-4485 for a possible
- // deadlock scenario that could have happened if continue to hold
- // the lock.
- this.lock.writeLock().unlock();
- }
+ private boolean completeFlush(List<HStoreFile> sfs, long snapshotId) throws IOException {
+ storeEngine.addStoreFiles(sfs);
// We do not need to call clearSnapshot method inside the write lock.
// The clearSnapshot itself is thread safe, which can be called at the same time with other
// memstore operations expect snapshot and clearSnapshot. And for these two methods, in HRegion
@@ -1259,11 +908,11 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
private void notifyChangedReadersObservers(List<HStoreFile> sfs) throws IOException {
for (ChangedReadersObserver o : this.changedReaderObservers) {
List<KeyValueScanner> memStoreScanners;
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
try {
memStoreScanners = this.memstore.getScanners(o.getReadPoint());
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
o.updateReaders(sfs, memStoreScanners);
}
@@ -1305,13 +954,13 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
byte[] stopRow, boolean includeStopRow, long readPt) throws IOException {
Collection<HStoreFile> storeFilesToScan;
List<KeyValueScanner> memStoreScanners;
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
try {
storeFilesToScan = this.storeEngine.getStoreFileManager().getFilesForScan(startRow,
includeStartRow, stopRow, includeStopRow);
memStoreScanners = this.memstore.getScanners(readPt);
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
try {
@@ -1388,11 +1037,11 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
boolean includeMemstoreScanner) throws IOException {
List<KeyValueScanner> memStoreScanners = null;
if (includeMemstoreScanner) {
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
try {
memStoreScanners = this.memstore.getScanners(readPt);
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
}
try {
@@ -1508,14 +1157,13 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
List<Path> newFiles) throws IOException {
// Do the steps necessary to complete the compaction.
setStoragePolicyFromFileName(newFiles);
- List<HStoreFile> sfs = commitStoreFiles(newFiles, true);
+ List<HStoreFile> sfs = storeEngine.commitStoreFiles(newFiles, true);
if (this.getCoprocessorHost() != null) {
for (HStoreFile sf : sfs) {
getCoprocessorHost().postCompact(this, sf, cr.getTracker(), cr, user);
}
}
- writeCompactionWalRecord(filesToCompact, sfs);
- replaceStoreFiles(filesToCompact, sfs);
+ replaceStoreFiles(filesToCompact, sfs, true);
if (cr.isMajor()) {
majorCompactedCellsCount.addAndGet(getCompactionProgress().getTotalCompactingKVs());
majorCompactedCellsSize.addAndGet(getCompactionProgress().totalCompactedSize);
@@ -1579,25 +1227,24 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
this.region.getRegionInfo(), compactionDescriptor, this.region.getMVCC());
}
- void replaceStoreFiles(Collection<HStoreFile> compactedFiles, Collection<HStoreFile> result)
- throws IOException {
- this.lock.writeLock().lock();
- try {
- this.storeEngine.getStoreFileManager().addCompactionResults(compactedFiles, result);
- synchronized (filesCompacting) {
- filesCompacting.removeAll(compactedFiles);
- }
-
- // These may be null when the RS is shutting down. The space quota Chores will fix the Region
- // sizes later so it's not super-critical if we miss these.
- RegionServerServices rsServices = region.getRegionServerServices();
- if (rsServices != null && rsServices.getRegionServerSpaceQuotaManager() != null) {
- updateSpaceQuotaAfterFileReplacement(
- rsServices.getRegionServerSpaceQuotaManager().getRegionSizeStore(), getRegionInfo(),
- compactedFiles, result);
- }
- } finally {
- this.lock.writeLock().unlock();
+ @RestrictedApi(explanation = "Should only be called in TestHStore", link = "",
+ allowedOnPath = ".*/(HStore|TestHStore).java")
+ void replaceStoreFiles(Collection<HStoreFile> compactedFiles, Collection<HStoreFile> result,
+ boolean writeCompactionMarker) throws IOException {
+ storeEngine.replaceStoreFiles(compactedFiles, result);
+ if (writeCompactionMarker) {
+ writeCompactionWalRecord(compactedFiles, result);
+ }
+ synchronized (filesCompacting) {
+ filesCompacting.removeAll(compactedFiles);
+ }
+ // These may be null when the RS is shutting down. The space quota Chores will fix the Region
+ // sizes later so it's not super-critical if we miss these.
+ RegionServerServices rsServices = region.getRegionServerServices();
+ if (rsServices != null && rsServices.getRegionServerSpaceQuotaManager() != null) {
+ updateSpaceQuotaAfterFileReplacement(
+ rsServices.getRegionServerSpaceQuotaManager().getRegionSizeStore(), getRegionInfo(),
+ compactedFiles, result);
}
}
@@ -1720,7 +1367,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
for (String compactionOutput : compactionOutputs) {
StoreFileInfo storeFileInfo =
getRegionFileSystem().getStoreFileInfo(getColumnFamilyName(), compactionOutput);
- HStoreFile storeFile = createStoreFileAndReader(storeFileInfo);
+ HStoreFile storeFile = storeEngine.createStoreFileAndReader(storeFileInfo);
outputStoreFiles.add(storeFile);
}
}
@@ -1728,7 +1375,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
if (!inputStoreFiles.isEmpty() || !outputStoreFiles.isEmpty()) {
LOG.info("Replaying compaction marker, replacing input files: " +
inputStoreFiles + " with output files : " + outputStoreFiles);
- this.replaceStoreFiles(inputStoreFiles, outputStoreFiles);
+ this.replaceStoreFiles(inputStoreFiles, outputStoreFiles, false);
this.refreshStoreSizeAndTotalBytes();
}
}
@@ -1737,14 +1384,14 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
public boolean hasReferences() {
// Grab the read lock here, because we need to ensure that: only when the atomic
// replaceStoreFiles(..) finished, we can get all the complete store file list.
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
try {
// Merge the current store files with compacted files here due to HBASE-20940.
Collection<HStoreFile> allStoreFiles = new ArrayList<>(getStorefiles());
allStoreFiles.addAll(getCompactedFiles());
return StoreUtils.hasReferences(allStoreFiles);
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
}
@@ -1784,7 +1431,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
final CompactionContext compaction = storeEngine.createCompaction();
CompactionRequestImpl request = null;
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
try {
synchronized (filesCompacting) {
// First, see if coprocessor would want to override selection.
@@ -1857,7 +1504,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
request.setTracker(tracker);
}
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
if (LOG.isDebugEnabled()) {
@@ -1890,7 +1537,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
this, getColumnFamilyDescriptor().getMinVersions());
return;
}
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
Collection<HStoreFile> delSfs = null;
try {
synchronized (filesCompacting) {
@@ -1902,7 +1549,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
}
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
if (CollectionUtils.isEmpty(delSfs)) {
@@ -1910,8 +1557,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
Collection<HStoreFile> newFiles = Collections.emptyList(); // No new files.
- writeCompactionWalRecord(delSfs, newFiles);
- replaceStoreFiles(delSfs, newFiles);
+ replaceStoreFiles(delSfs, newFiles, true);
refreshStoreSizeAndTotalBytes();
LOG.info("Completed removal of " + delSfs.size() + " unnecessary (expired) file(s) in "
+ this + "; total size is "
@@ -1934,25 +1580,6 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
/**
- * Validates a store file by opening and closing it. In HFileV2 this should not be an expensive
- * operation.
- * @param path the path to the store file
- */
- private void validateStoreFile(Path path) throws IOException {
- HStoreFile storeFile = null;
- try {
- storeFile = createStoreFileAndReader(path);
- } catch (IOException e) {
- LOG.error("Failed to open store file : {}, keeping it in tmp location", path, e);
- throw e;
- } finally {
- if (storeFile != null) {
- storeFile.closeStoreFile(false);
- }
- }
- }
-
- /**
* Update counts.
*/
protected void refreshStoreSizeAndTotalBytes()
@@ -1997,7 +1624,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
* Determines if Store should be split.
*/
public Optional<byte[]> getSplitPoint() {
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
try {
// Should already be enforced by the split policy!
assert !this.getRegionInfo().isMetaRegion();
@@ -2010,7 +1637,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
} catch(IOException e) {
LOG.warn("Failed getting store size for {}", this, e);
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
return Optional.empty();
}
@@ -2043,7 +1670,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
*/
public KeyValueScanner getScanner(Scan scan, final NavigableSet<byte[]> targetCols, long readPt)
throws IOException {
- lock.readLock().lock();
+ storeEngine.readLock();
try {
ScanInfo scanInfo;
if (this.getCoprocessorHost() != null) {
@@ -2053,7 +1680,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
return createScanner(scan, scanInfo, targetCols, readPt);
} finally {
- lock.readLock().unlock();
+ storeEngine.readUnlock();
}
}
@@ -2083,7 +1710,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
boolean cacheBlocks, boolean usePread, boolean isCompaction, ScanQueryMatcher matcher,
byte[] startRow, boolean includeStartRow, byte[] stopRow, boolean includeStopRow, long readPt,
boolean includeMemstoreScanner) throws IOException {
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
try {
Map<String, HStoreFile> name2File =
new HashMap<>(getStorefilesCount() + getCompactedFilesCount());
@@ -2108,7 +1735,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
return getScanners(filesToReopen, cacheBlocks, false, false, matcher, startRow,
includeStartRow, stopRow, includeStopRow, readPt, false);
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
}
@@ -2174,41 +1801,20 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
@Override
public long getStorefilesSize() {
// Include all StoreFiles
- return getStorefilesSize(this.storeEngine.getStoreFileManager().getStorefiles(), sf -> true);
+ return StoreUtils.getStorefilesSize(this.storeEngine.getStoreFileManager().getStorefiles(),
+ sf -> true);
}
@Override
public long getHFilesSize() {
// Include only StoreFiles which are HFiles
- return getStorefilesSize(this.storeEngine.getStoreFileManager().getStorefiles(),
+ return StoreUtils.getStorefilesSize(this.storeEngine.getStoreFileManager().getStorefiles(),
HStoreFile::isHFile);
}
- private long getTotalUncompressedBytes(List<HStoreFile> files) {
- return files.stream()
- .mapToLong(file -> getStorefileFieldSize(file, StoreFileReader::getTotalUncompressedBytes))
- .sum();
- }
-
- private long getStorefilesSize(Collection<HStoreFile> files, Predicate<HStoreFile> predicate) {
- return files.stream().filter(predicate)
- .mapToLong(file -> getStorefileFieldSize(file, StoreFileReader::length)).sum();
- }
-
- private long getStorefileFieldSize(HStoreFile file, ToLongFunction<StoreFileReader> f) {
- if (file == null) {
- return 0L;
- }
- StoreFileReader reader = file.getReader();
- if (reader == null) {
- return 0L;
- }
- return f.applyAsLong(reader);
- }
-
private long getStorefilesFieldSize(ToLongFunction<StoreFileReader> f) {
return this.storeEngine.getStoreFileManager().getStorefiles().stream()
- .mapToLong(file -> getStorefileFieldSize(file, f)).sum();
+ .mapToLong(file -> StoreUtils.getStorefileFieldSize(file, f)).sum();
}
@Override
@@ -2279,11 +1885,11 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
*/
public void upsert(Iterable<Cell> cells, long readpoint, MemStoreSizing memstoreSizing)
throws IOException {
- this.lock.readLock().lock();
+ this.storeEngine.readLock();
try {
this.memstore.upsert(cells, readpoint, memstoreSizing);
} finally {
- this.lock.readLock().unlock();
+ this.storeEngine.readUnlock();
}
}
@@ -2336,7 +1942,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
return false;
}
status.setStatus("Flushing " + this + ": reopening flushed file");
- List<HStoreFile> storeFiles = commitStoreFiles(tempFiles, false);
+ List<HStoreFile> storeFiles = storeEngine.commitStoreFiles(tempFiles, false);
for (HStoreFile sf : storeFiles) {
StoreFileReader r = sf.getReader();
if (LOG.isInfoEnabled()) {
@@ -2359,7 +1965,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
}
}
// Add new file to store files. Clear snapshot too while we have the Store write lock.
- return updateStorefiles(storeFiles, snapshot.getId());
+ return completeFlush(storeFiles, snapshot.getId());
}
@Override
@@ -2387,7 +1993,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
// open the file as a store file (hfile link, etc)
StoreFileInfo storeFileInfo =
getRegionFileSystem().getStoreFileInfo(getColumnFamilyName(), file);
- HStoreFile storeFile = createStoreFileAndReader(storeFileInfo);
+ HStoreFile storeFile = storeEngine.createStoreFileAndReader(storeFileInfo);
storeFiles.add(storeFile);
HStore.this.storeSize.addAndGet(storeFile.getReader().length());
HStore.this.totalUncompressedBytes
@@ -2404,7 +2010,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
snapshotId = snapshot.getId();
snapshot.close();
}
- HStore.this.updateStorefiles(storeFiles, snapshotId);
+ HStore.this.completeFlush(storeFiles, snapshotId);
}
/**
@@ -2417,7 +2023,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
//won't be closed. If we are using MSLAB, the chunk referenced by those scanners
//can't be released, thus memory leak
snapshot.close();
- HStore.this.updateStorefiles(Collections.emptyList(), snapshot.getId());
+ HStore.this.completeFlush(Collections.emptyList(), snapshot.getId());
}
}
}
@@ -2583,7 +2189,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
// ensure other threads do not attempt to archive the same files on close()
archiveLock.lock();
try {
- lock.readLock().lock();
+ storeEngine.readLock();
Collection<HStoreFile> copyCompactedfiles = null;
try {
Collection<HStoreFile> compactedfiles =
@@ -2595,7 +2201,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
LOG.trace("No compacted files to archive");
}
} finally {
- lock.readLock().unlock();
+ storeEngine.readUnlock();
}
if (CollectionUtils.isNotEmpty(copyCompactedfiles)) {
removeCompactedfiles(copyCompactedfiles, true);
@@ -2730,12 +2336,7 @@ public class HStore implements Store, HeapSize, StoreConfigInformation,
private void clearCompactedfiles(List<HStoreFile> filesToRemove) throws IOException {
LOG.trace("Clearing the compacted file {} from this store", filesToRemove);
- try {
- lock.writeLock().lock();
- this.getStoreEngine().getStoreFileManager().removeCompactedFiles(filesToRemove);
- } finally {
- lock.writeLock().unlock();
- }
+ storeEngine.removeCompactedFiles(filesToRemove);
}
void reportArchivedFilesForQuota(List<? extends StoreFile> archivedFiles, List<Long> fileSizes) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreContext.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreContext.java
index 2623350..2a9f968 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreContext.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreContext.java
@@ -23,6 +23,7 @@ import java.util.function.Supplier;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.io.HeapSize;
import org.apache.hadoop.hbase.io.crypto.Encryption;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
@@ -108,6 +109,14 @@ public final class StoreContext implements HeapSize {
return coprocessorHost;
}
+ public RegionInfo getRegionInfo() {
+ return regionFileSystem.getRegionInfo();
+ }
+
+ public boolean isPrimaryReplicaStore() {
+ return getRegionInfo().getReplicaId() == RegionInfo.DEFAULT_REPLICA_ID;
+ }
+
public static Builder getBuilder() {
return new Builder();
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java
index 60b3c3d..f9d6c29 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java
@@ -20,37 +20,129 @@
package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
-
+import java.util.Set;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CellComparator;
+import org.apache.hadoop.hbase.log.HBaseMarkers;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionPolicy;
import org.apache.hadoop.hbase.regionserver.compactions.Compactor;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
+import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;
/**
- * StoreEngine is a factory that can create the objects necessary for HStore to operate.
- * Since not all compaction policies, compactors and store file managers are compatible,
- * they are tied together and replaced together via StoreEngine-s.
+ * StoreEngine is a factory that can create the objects necessary for HStore to operate. Since not
+ * all compaction policies, compactors and store file managers are compatible, they are tied
+ * together and replaced together via StoreEngine-s.
+ * <p/>
+ * We expose read write lock methods to upper layer for store operations:<br/>
+ * <ul>
+ * <li>Locked in shared mode when the list of component stores is looked at:
+ * <ul>
+ * <li>all reads/writes to table data</li>
+ * <li>checking for split</li>
+ * </ul>
+ * </li>
+ * <li>Locked in exclusive mode when the list of component stores is modified:
+ * <ul>
+ * <li>closing</li>
+ * <li>completing a compaction</li>
+ * </ul>
+ * </li>
+ * </ul>
+ * <p/>
+ * It is a bit confusing that we have a StoreFileManager(SFM) and then a StoreFileTracker(SFT). As
+ * its name says, SFT is used to track the store files list. The reason why we have a SFT beside SFM
+ * is that, when introducing stripe compaction, we introduced the StoreEngine and also the SFM, but
+ * actually, the SFM here is not a general 'Manager', it is only designed to manage the in memory
+ * 'stripes', so we can select different store files when scanning or compacting. The 'tracking' of
+ * store files is actually done in {@link org.apache.hadoop.hbase.regionserver.HRegionFileSystem}
+ * and {@link HStore} before we have SFT. And since SFM is designed to only holds in memory states,
+ * we will hold write lock when updating it, the lock is also used to protect the normal read/write
+ * requests. This means we'd better not add IO operations to SFM. And also, no matter what the in
+ * memory state is, stripe or not, it does not effect how we track the store files. So consider all
+ * these facts, here we introduce a separated SFT to track the store files.
+ * <p/>
+ * Here, since we always need to update SFM and SFT almost at the same time, we introduce methods in
+ * StoreEngine directly to update them both, so upper layer just need to update StoreEngine once, to
+ * reduce the possible misuse.
*/
@InterfaceAudience.Private
-public abstract class StoreEngine<SF extends StoreFlusher,
- CP extends CompactionPolicy, C extends Compactor, SFM extends StoreFileManager> {
+public abstract class StoreEngine<SF extends StoreFlusher, CP extends CompactionPolicy,
+ C extends Compactor, SFM extends StoreFileManager> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StoreEngine.class);
+
protected SF storeFlusher;
protected CP compactionPolicy;
protected C compactor;
protected SFM storeFileManager;
+ private Configuration conf;
+ private StoreContext ctx;
+ private RegionCoprocessorHost coprocessorHost;
+ private Function<String, ExecutorService> openStoreFileThreadPoolCreator;
+ private StoreFileTracker storeFileTracker;
+
+ private final ReadWriteLock storeLock = new ReentrantReadWriteLock();
/**
- * The name of the configuration parameter that specifies the class of
- * a store engine that is used to manage and compact HBase store files.
+ * The name of the configuration parameter that specifies the class of a store engine that is used
+ * to manage and compact HBase store files.
*/
public static final String STORE_ENGINE_CLASS_KEY = "hbase.hstore.engine.class";
- private static final Class<? extends StoreEngine<?, ?, ?, ?>>
- DEFAULT_STORE_ENGINE_CLASS = DefaultStoreEngine.class;
+ private static final Class<? extends StoreEngine<?, ?, ?, ?>> DEFAULT_STORE_ENGINE_CLASS =
+ DefaultStoreEngine.class;
+
+ /**
+ * Acquire read lock of this store.
+ */
+ public void readLock() {
+ storeLock.readLock().lock();
+ }
+
+ /**
+ * Release read lock of this store.
+ */
+ public void readUnlock() {
+ storeLock.readLock().unlock();
+ }
+
+ /**
+ * Acquire write lock of this store.
+ */
+ public void writeLock() {
+ storeLock.writeLock().lock();
+ }
+
+ /**
+ * Release write lock of this store.
+ */
+ public void writeUnlock() {
+ storeLock.writeLock().unlock();
+ }
/**
* @return Compaction policy to use.
@@ -80,6 +172,11 @@ public abstract class StoreEngine<SF extends StoreFlusher,
return this.storeFlusher;
}
+ private StoreFileTracker createStoreFileTracker(HStore store) {
+ return StoreFileTrackerFactory.create(store.conf, store.getRegionInfo().getTable(),
+ store.isPrimaryReplicaStore(), store.getStoreContext());
+ }
+
/**
* @param filesCompacting Files currently compacting
* @return whether a compaction selection is possible
@@ -87,8 +184,8 @@ public abstract class StoreEngine<SF extends StoreFlusher,
public abstract boolean needsCompaction(List<HStoreFile> filesCompacting);
/**
- * Creates an instance of a compaction context specific to this engine.
- * Doesn't actually select or start a compaction. See CompactionContext class comment.
+ * Creates an instance of a compaction context specific to this engine. Doesn't actually select or
+ * start a compaction. See CompactionContext class comment.
* @return New CompactionContext object.
*/
public abstract CompactionContext createCompaction() throws IOException;
@@ -96,32 +193,321 @@ public abstract class StoreEngine<SF extends StoreFlusher,
/**
* Create the StoreEngine's components.
*/
- protected abstract void createComponents(
- Configuration conf, HStore store, CellComparator cellComparator) throws IOException;
+ protected abstract void createComponents(Configuration conf, HStore store,
+ CellComparator cellComparator) throws IOException;
- private void createComponentsOnce(
- Configuration conf, HStore store, CellComparator cellComparator) throws IOException {
- assert compactor == null && compactionPolicy == null
- && storeFileManager == null && storeFlusher == null;
+ protected final void createComponentsOnce(Configuration conf, HStore store,
+ CellComparator cellComparator) throws IOException {
+ assert compactor == null && compactionPolicy == null && storeFileManager == null &&
+ storeFlusher == null && storeFileTracker == null;
createComponents(conf, store, cellComparator);
- assert compactor != null && compactionPolicy != null
- && storeFileManager != null && storeFlusher != null;
+ this.conf = conf;
+ this.ctx = store.getStoreContext();
+ this.coprocessorHost = store.getHRegion().getCoprocessorHost();
+ this.openStoreFileThreadPoolCreator = store.getHRegion()::getStoreFileOpenAndCloseThreadPool;
+ this.storeFileTracker = createStoreFileTracker(store);
+ assert compactor != null && compactionPolicy != null && storeFileManager != null &&
+ storeFlusher != null && storeFileTracker != null;
+ }
+
+ /**
+ * Create a writer for writing new store files.
+ * @return Writer for a new StoreFile
+ */
+ public StoreFileWriter createWriter(CreateStoreFileWriterParams params) throws IOException {
+ return storeFileTracker.createWriter(params);
+ }
+
+ public HStoreFile createStoreFileAndReader(Path p) throws IOException {
+ StoreFileInfo info = new StoreFileInfo(conf, ctx.getRegionFileSystem().getFileSystem(), p,
+ ctx.isPrimaryReplicaStore());
+ return createStoreFileAndReader(info);
+ }
+
+ public HStoreFile createStoreFileAndReader(StoreFileInfo info) throws IOException {
+ info.setRegionCoprocessorHost(coprocessorHost);
+ HStoreFile storeFile =
+ new HStoreFile(info, ctx.getFamily().getBloomFilterType(), ctx.getCacheConf());
+ storeFile.initReader();
+ return storeFile;
+ }
+
+ /**
+ * Validates a store file by opening and closing it. In HFileV2 this should not be an expensive
+ * operation.
+ * @param path the path to the store file
+ */
+ public void validateStoreFile(Path path) throws IOException {
+ HStoreFile storeFile = null;
+ try {
+ storeFile = createStoreFileAndReader(path);
+ } catch (IOException e) {
+ LOG.error("Failed to open store file : {}, keeping it in tmp location", path, e);
+ throw e;
+ } finally {
+ if (storeFile != null) {
+ storeFile.closeStoreFile(false);
+ }
+ }
+ }
+
+ private List<HStoreFile> openStoreFiles(Collection<StoreFileInfo> files, boolean warmup)
+ throws IOException {
+ if (CollectionUtils.isEmpty(files)) {
+ return Collections.emptyList();
+ }
+ // initialize the thread pool for opening store files in parallel..
+ ExecutorService storeFileOpenerThreadPool =
+ openStoreFileThreadPoolCreator.apply("StoreFileOpener-" +
+ ctx.getRegionInfo().getEncodedName() + "-" + ctx.getFamily().getNameAsString());
+ CompletionService<HStoreFile> completionService =
+ new ExecutorCompletionService<>(storeFileOpenerThreadPool);
+
+ int totalValidStoreFile = 0;
+ for (StoreFileInfo storeFileInfo : files) {
+ // open each store file in parallel
+ completionService.submit(() -> createStoreFileAndReader(storeFileInfo));
+ totalValidStoreFile++;
+ }
+
+ Set<String> compactedStoreFiles = new HashSet<>();
+ ArrayList<HStoreFile> results = new ArrayList<>(files.size());
+ IOException ioe = null;
+ try {
+ for (int i = 0; i < totalValidStoreFile; i++) {
+ try {
+ HStoreFile storeFile = completionService.take().get();
+ if (storeFile != null) {
+ LOG.debug("loaded {}", storeFile);
+ results.add(storeFile);
+ compactedStoreFiles.addAll(storeFile.getCompactedStoreFiles());
+ }
+ } catch (InterruptedException e) {
+ if (ioe == null) {
+ ioe = new InterruptedIOException(e.getMessage());
+ }
+ } catch (ExecutionException e) {
+ if (ioe == null) {
+ ioe = new IOException(e.getCause());
+ }
+ }
+ }
+ } finally {
+ storeFileOpenerThreadPool.shutdownNow();
+ }
+ if (ioe != null) {
+ // close StoreFile readers
+ boolean evictOnClose =
+ ctx.getCacheConf() != null ? ctx.getCacheConf().shouldEvictOnClose() : true;
+ for (HStoreFile file : results) {
+ try {
+ if (file != null) {
+ file.closeStoreFile(evictOnClose);
+ }
+ } catch (IOException e) {
+ LOG.warn("Could not close store file {}", file, e);
+ }
+ }
+ throw ioe;
+ }
+
+ // Should not archive the compacted store files when region warmup. See HBASE-22163.
+ if (!warmup) {
+ // Remove the compacted files from result
+ List<HStoreFile> filesToRemove = new ArrayList<>(compactedStoreFiles.size());
+ for (HStoreFile storeFile : results) {
+ if (compactedStoreFiles.contains(storeFile.getPath().getName())) {
+ LOG.warn("Clearing the compacted storefile {} from {}", storeFile, this);
+ storeFile.getReader().close(
+ storeFile.getCacheConf() != null ? storeFile.getCacheConf().shouldEvictOnClose() :
+ true);
+ filesToRemove.add(storeFile);
+ }
+ }
+ results.removeAll(filesToRemove);
+ if (!filesToRemove.isEmpty() && ctx.isPrimaryReplicaStore()) {
+ LOG.debug("Moving the files {} to archive", filesToRemove);
+ ctx.getRegionFileSystem().removeStoreFiles(ctx.getFamily().getNameAsString(),
+ filesToRemove);
+ }
+ }
+
+ return results;
+ }
+
+ public void initialize(boolean warmup) throws IOException {
+ List<StoreFileInfo> fileInfos = storeFileTracker.load();
+ List<HStoreFile> files = openStoreFiles(fileInfos, warmup);
+ storeFileManager.loadFiles(files);
+ }
+
+ public void refreshStoreFiles() throws IOException {
+ List<StoreFileInfo> fileInfos = storeFileTracker.load();
+ refreshStoreFilesInternal(fileInfos);
+ }
+
+ public void refreshStoreFiles(Collection<String> newFiles) throws IOException {
+ List<StoreFileInfo> storeFiles = new ArrayList<>(newFiles.size());
+ for (String file : newFiles) {
+ storeFiles
+ .add(ctx.getRegionFileSystem().getStoreFileInfo(ctx.getFamily().getNameAsString(), file));
+ }
+ refreshStoreFilesInternal(storeFiles);
+ }
+
+ /**
+ * Checks the underlying store files, and opens the files that have not been opened, and removes
+ * the store file readers for store files no longer available. Mainly used by secondary region
+ * replicas to keep up to date with the primary region files.
+ */
+ private void refreshStoreFilesInternal(Collection<StoreFileInfo> newFiles) throws IOException {
+ Collection<HStoreFile> currentFiles = storeFileManager.getStorefiles();
+ Collection<HStoreFile> compactedFiles = storeFileManager.getCompactedfiles();
+ if (currentFiles == null) {
+ currentFiles = Collections.emptySet();
+ }
+ if (newFiles == null) {
+ newFiles = Collections.emptySet();
+ }
+ if (compactedFiles == null) {
+ compactedFiles = Collections.emptySet();
+ }
+
+ HashMap<StoreFileInfo, HStoreFile> currentFilesSet = new HashMap<>(currentFiles.size());
+ for (HStoreFile sf : currentFiles) {
+ currentFilesSet.put(sf.getFileInfo(), sf);
+ }
+ HashMap<StoreFileInfo, HStoreFile> compactedFilesSet = new HashMap<>(compactedFiles.size());
+ for (HStoreFile sf : compactedFiles) {
+ compactedFilesSet.put(sf.getFileInfo(), sf);
+ }
+
+ Set<StoreFileInfo> newFilesSet = new HashSet<StoreFileInfo>(newFiles);
+ // Exclude the files that have already been compacted
+ newFilesSet = Sets.difference(newFilesSet, compactedFilesSet.keySet());
+ Set<StoreFileInfo> toBeAddedFiles = Sets.difference(newFilesSet, currentFilesSet.keySet());
+ Set<StoreFileInfo> toBeRemovedFiles = Sets.difference(currentFilesSet.keySet(), newFilesSet);
+
+ if (toBeAddedFiles.isEmpty() && toBeRemovedFiles.isEmpty()) {
+ return;
+ }
+
+ LOG.info("Refreshing store files for " + this + " files to add: " + toBeAddedFiles +
+ " files to remove: " + toBeRemovedFiles);
+
+ Set<HStoreFile> toBeRemovedStoreFiles = new HashSet<>(toBeRemovedFiles.size());
+ for (StoreFileInfo sfi : toBeRemovedFiles) {
+ toBeRemovedStoreFiles.add(currentFilesSet.get(sfi));
+ }
+
+ // try to open the files
+ List<HStoreFile> openedFiles = openStoreFiles(toBeAddedFiles, false);
+
+ // propogate the file changes to the underlying store file manager
+ replaceStoreFiles(toBeRemovedStoreFiles, openedFiles); // won't throw an exception
+ }
+
+ /**
+ * Commit the given {@code files}.
+ * <p/>
+ * We will move the file into data directory, and open it.
+ * @param files the files want to commit
+ * @param validate whether to validate the store files
+ * @return the committed store files
+ */
+ public List<HStoreFile> commitStoreFiles(List<Path> files, boolean validate) throws IOException {
+ List<HStoreFile> committedFiles = new ArrayList<>(files.size());
+ HRegionFileSystem hfs = ctx.getRegionFileSystem();
+ String familyName = ctx.getFamily().getNameAsString();
+ Path storeDir = hfs.getStoreDir(familyName);
+ for (Path file : files) {
+ try {
+ if (validate) {
+ validateStoreFile(file);
+ }
+ Path committedPath;
+ // As we want to support writing to data directory directly, here we need to check whether
+ // the store file is already in the right place
+ if (file.getParent() != null && file.getParent().equals(storeDir)) {
+ // already in the right place, skip renmaing
+ committedPath = file;
+ } else {
+ // Write-out finished successfully, move into the right spot
+ committedPath = hfs.commitStoreFile(familyName, file);
+ }
+ HStoreFile sf = createStoreFileAndReader(committedPath);
+ committedFiles.add(sf);
+ } catch (IOException e) {
+ LOG.error("Failed to commit store file {}", file, e);
+ // Try to delete the files we have committed before.
+ // It is OK to fail when deleting as leaving the file there does not cause any data
+ // corruption problem. It just introduces some duplicated data which may impact read
+ // performance a little when reading before compaction.
+ for (HStoreFile sf : committedFiles) {
+ Path pathToDelete = sf.getPath();
+ try {
+ sf.deleteStoreFile();
+ } catch (IOException deleteEx) {
+ LOG.warn(HBaseMarkers.FATAL, "Failed to delete committed store file {}", pathToDelete,
+ deleteEx);
+ }
+ }
+ throw new IOException("Failed to commit the flush", e);
+ }
+ }
+ return committedFiles;
+ }
+
+ public void addStoreFiles(Collection<HStoreFile> storeFiles) throws IOException {
+ storeFileTracker.add(StoreUtils.toStoreFileInfo(storeFiles));
+ writeLock();
+ try {
+ storeFileManager.insertNewFiles(storeFiles);
+ } finally {
+ // We need the lock, as long as we are updating the storeFiles
+ // or changing the memstore. Let us release it before calling
+ // notifyChangeReadersObservers. See HBASE-4485 for a possible
+ // deadlock scenario that could have happened if continue to hold
+ // the lock.
+ writeUnlock();
+ }
+ }
+
+ public void replaceStoreFiles(Collection<HStoreFile> compactedFiles,
+ Collection<HStoreFile> newFiles) throws IOException {
+ storeFileTracker.replace(StoreUtils.toStoreFileInfo(compactedFiles),
+ StoreUtils.toStoreFileInfo(newFiles));
+ writeLock();
+ try {
+ storeFileManager.addCompactionResults(compactedFiles, newFiles);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ public void removeCompactedFiles(Collection<HStoreFile> compactedFiles) {
+ writeLock();
+ try {
+ storeFileManager.removeCompactedFiles(compactedFiles);
+ } finally {
+ writeUnlock();
+ }
}
/**
* Create the StoreEngine configured for the given Store.
- * @param store The store. An unfortunate dependency needed due to it
- * being passed to coprocessors via the compactor.
+ * @param store The store. An unfortunate dependency needed due to it being passed to coprocessors
+ * via the compactor.
* @param conf Store configuration.
* @param cellComparator CellComparator for storeFileManager.
* @return StoreEngine to use.
*/
- public static StoreEngine<?, ?, ?, ?> create(
- HStore store, Configuration conf, CellComparator cellComparator) throws IOException {
+ public static StoreEngine<?, ?, ?, ?> create(HStore store, Configuration conf,
+ CellComparator cellComparator) throws IOException {
String className = conf.get(STORE_ENGINE_CLASS_KEY, DEFAULT_STORE_ENGINE_CLASS.getName());
try {
- StoreEngine<?,?,?,?> se = ReflectionUtils.instantiateWithCustomCtor(
- className, new Class[] { }, new Object[] { });
+ StoreEngine<?, ?, ?, ?> se =
+ ReflectionUtils.instantiateWithCustomCtor(className, new Class[] {}, new Object[] {});
se.createComponentsOnce(conf, store, cellComparator);
return se;
} catch (Exception e) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileManager.java
index 27127f3..a40b209 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileManager.java
@@ -18,6 +18,7 @@
*/
package org.apache.hadoop.hbase.regionserver;
+import com.google.errorprone.annotations.RestrictedApi;
import java.io.IOException;
import java.util.Collection;
import java.util.Comparator;
@@ -49,12 +50,16 @@ public interface StoreFileManager {
* Loads the initial store files into empty StoreFileManager.
* @param storeFiles The files to load.
*/
+ @RestrictedApi(explanation = "Should only be called in StoreEngine", link = "",
+ allowedOnPath = ".*(/org/apache/hadoop/hbase/regionserver/StoreEngine.java|/src/test/.*)")
void loadFiles(List<HStoreFile> storeFiles);
/**
* Adds new files, either for from MemStore flush or bulk insert, into the structure.
* @param sfs New store files.
*/
+ @RestrictedApi(explanation = "Should only be called in StoreEngine", link = "",
+ allowedOnPath = ".*(/org/apache/hadoop/hbase/regionserver/StoreEngine.java|/src/test/.*)")
void insertNewFiles(Collection<HStoreFile> sfs);
/**
@@ -62,12 +67,16 @@ public interface StoreFileManager {
* @param compactedFiles The input files for the compaction.
* @param results The resulting files for the compaction.
*/
+ @RestrictedApi(explanation = "Should only be called in StoreEngine", link = "",
+ allowedOnPath = ".*(/org/apache/hadoop/hbase/regionserver/StoreEngine.java|/src/test/.*)")
void addCompactionResults(Collection<HStoreFile> compactedFiles, Collection<HStoreFile> results);
/**
* Remove the compacted files
* @param compactedFiles the list of compacted files
*/
+ @RestrictedApi(explanation = "Should only be called in StoreEngine", link = "",
+ allowedOnPath = ".*(/org/apache/hadoop/hbase/regionserver/StoreEngine.java|/src/test/.*)")
void removeCompactedFiles(Collection<HStoreFile> compactedFiles);
/**
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java
index 1064b6c..5803128 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFlusher.java
@@ -70,10 +70,17 @@ abstract class StoreFlusher {
writer.close();
}
+ protected final StoreFileWriter createWriter(MemStoreSnapshot snapshot, boolean alwaysIncludesTag)
+ throws IOException {
+ return store.getStoreEngine()
+ .createWriter(CreateStoreFileWriterParams.create().maxKeyCount(snapshot.getCellsCount())
+ .compression(store.getColumnFamilyDescriptor().getCompressionType()).isCompaction(false)
+ .includeMVCCReadpoint(true).includesTag(alwaysIncludesTag || snapshot.isTagsPresent())
+ .shouldDropBehind(false));
+ }
/**
* Creates the scanner for flushing snapshot. Also calls coprocessors.
- * @param snapshotScanners
* @return The scanner; null if coprocessor is canceling the flush.
*/
protected final InternalScanner createScanner(List<KeyValueScanner> snapshotScanners,
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreUtils.java
index ac5955f..e464b86 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreUtils.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreUtils.java
@@ -20,10 +20,13 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.Collection;
+import java.util.List;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
-
+import java.util.function.Predicate;
+import java.util.function.ToLongFunction;
+import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellComparator;
@@ -39,10 +42,13 @@ import org.slf4j.LoggerFactory;
* Utility functions for region server storage layer.
*/
@InterfaceAudience.Private
-public class StoreUtils {
+public final class StoreUtils {
private static final Logger LOG = LoggerFactory.getLogger(StoreUtils.class);
+ private StoreUtils() {
+ }
+
/**
* Creates a deterministic hash code for store file collection.
*/
@@ -161,4 +167,30 @@ public class StoreUtils {
HFile.DEFAULT_BYTES_PER_CHECKSUM);
}
+ public static List<StoreFileInfo> toStoreFileInfo(Collection<HStoreFile> storefiles) {
+ return storefiles.stream().map(HStoreFile::getFileInfo).collect(Collectors.toList());
+ }
+
+ public static long getTotalUncompressedBytes(List<HStoreFile> files) {
+ return files.stream()
+ .mapToLong(file -> getStorefileFieldSize(file, StoreFileReader::getTotalUncompressedBytes))
+ .sum();
+ }
+
+ public static long getStorefilesSize(Collection<HStoreFile> files,
+ Predicate<HStoreFile> predicate) {
+ return files.stream().filter(predicate)
+ .mapToLong(file -> getStorefileFieldSize(file, StoreFileReader::length)).sum();
+ }
+
+ public static long getStorefileFieldSize(HStoreFile file, ToLongFunction<StoreFileReader> f) {
+ if (file == null) {
+ return 0L;
+ }
+ StoreFileReader reader = file.getReader();
+ if (reader == null) {
+ return 0L;
+ }
+ return f.applyAsLong(reader);
+ }
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreEngine.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreEngine.java
index 14863a6..bfb3f64 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreEngine.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreEngine.java
@@ -20,20 +20,19 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.HBaseInterfaceAudience;
-import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
-import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
import org.apache.hadoop.hbase.regionserver.compactions.StripeCompactionPolicy;
import org.apache.hadoop.hbase.regionserver.compactions.StripeCompactor;
import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
import org.apache.hadoop.hbase.security.User;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreFlusher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreFlusher.java
index 1560aef..f8183b7 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreFlusher.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StripeStoreFlusher.java
@@ -70,7 +70,7 @@ public class StripeStoreFlusher extends StoreFlusher {
StripeMultiFileWriter mw = null;
try {
mw = req.createWriter(); // Writer according to the policy.
- StripeMultiFileWriter.WriterFactory factory = createWriterFactory(cellsCount);
+ StripeMultiFileWriter.WriterFactory factory = createWriterFactory(snapshot);
StoreScanner storeScanner = (scanner instanceof StoreScanner) ? (StoreScanner)scanner : null;
mw.init(storeScanner, factory);
@@ -98,13 +98,12 @@ public class StripeStoreFlusher extends StoreFlusher {
return result;
}
- private StripeMultiFileWriter.WriterFactory createWriterFactory(final long kvCount) {
+ private StripeMultiFileWriter.WriterFactory createWriterFactory(MemStoreSnapshot snapshot) {
return new StripeMultiFileWriter.WriterFactory() {
@Override
public StoreFileWriter createWriter() throws IOException {
- StoreFileWriter writer = store.createWriterInTmp(kvCount,
- store.getColumnFamilyDescriptor().getCompressionType(), false, true, true, false);
- return writer;
+ // XXX: it used to always pass true for includesTag, re-consider?
+ return StripeStoreFlusher.this.createWriter(snapshot, true);
}
};
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/AbstractMultiOutputCompactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/AbstractMultiOutputCompactor.java
index 42841bf..533be17 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/AbstractMultiOutputCompactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/AbstractMultiOutputCompactor.java
@@ -51,13 +51,14 @@ public abstract class AbstractMultiOutputCompactor<T extends AbstractMultiFileWr
WriterFactory writerFactory = new WriterFactory() {
@Override
public StoreFileWriter createWriter() throws IOException {
- return createTmpWriter(fd, shouldDropBehind, major);
+ return AbstractMultiOutputCompactor.this.createWriter(fd, shouldDropBehind, major);
}
@Override
public StoreFileWriter createWriterWithStoragePolicy(String fileStoragePolicy)
- throws IOException {
- return createTmpWriter(fd, shouldDropBehind, fileStoragePolicy, major);
+ throws IOException {
+ return AbstractMultiOutputCompactor.this.createWriter(fd, shouldDropBehind,
+ fileStoragePolicy, major);
}
};
// Prepare multi-writer, and perform the compaction using scanner and writer.
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
index e524f7d..47ef0f2 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
@@ -28,7 +28,6 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
@@ -39,6 +38,7 @@ import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileInfo;
import org.apache.hadoop.hbase.regionserver.CellSink;
+import org.apache.hadoop.hbase.regionserver.CreateStoreFileWriterParams;
import org.apache.hadoop.hbase.regionserver.HStore;
import org.apache.hadoop.hbase.regionserver.HStoreFile;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
@@ -61,6 +61,7 @@ import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
/**
@@ -261,29 +262,32 @@ public abstract class Compactor<T extends CellSink> {
}
};
+ protected final CreateStoreFileWriterParams createParams(FileDetails fd, boolean shouldDropBehind,
+ boolean major) {
+ return CreateStoreFileWriterParams.create().maxKeyCount(fd.maxKeyCount)
+ .compression(major ? majorCompactionCompression : minorCompactionCompression)
+ .isCompaction(true).includeMVCCReadpoint(fd.maxMVCCReadpoint > 0)
+ .includesTag(fd.maxTagsLength > 0).shouldDropBehind(shouldDropBehind)
+ .totalCompactedFilesSize(fd.totalCompactedFilesSize);
+ }
+
/**
- * Creates a writer for a new file in a temporary directory.
+ * Creates a writer for a new file.
* @param fd The file details.
- * @return Writer for a new StoreFile in the tmp dir.
+ * @return Writer for a new StoreFile
* @throws IOException if creation failed
*/
- protected final StoreFileWriter createTmpWriter(FileDetails fd, boolean shouldDropBehind, boolean major)
- throws IOException {
+ protected final StoreFileWriter createWriter(FileDetails fd, boolean shouldDropBehind,
+ boolean major) throws IOException {
// When all MVCC readpoints are 0, don't write them.
// See HBASE-8166, HBASE-12600, and HBASE-13389.
- return store.createWriterInTmp(fd.maxKeyCount,
- major ? majorCompactionCompression : minorCompactionCompression,
- true, fd.maxMVCCReadpoint > 0,
- fd.maxTagsLength > 0, shouldDropBehind, fd.totalCompactedFilesSize,
- HConstants.EMPTY_STRING);
+ return store.getStoreEngine().createWriter(createParams(fd, shouldDropBehind, major));
}
- protected final StoreFileWriter createTmpWriter(FileDetails fd, boolean shouldDropBehind,
- String fileStoragePolicy, boolean major) throws IOException {
- return store.createWriterInTmp(fd.maxKeyCount,
- major ? majorCompactionCompression : minorCompactionCompression,
- true, fd.maxMVCCReadpoint > 0,
- fd.maxTagsLength > 0, shouldDropBehind, fd.totalCompactedFilesSize, fileStoragePolicy);
+ protected final StoreFileWriter createWriter(FileDetails fd, boolean shouldDropBehind,
+ String fileStoragePolicy, boolean major) throws IOException {
+ return store.getStoreEngine()
+ .createWriter(createParams(fd, shouldDropBehind, major).fileStoragePolicy(fileStoragePolicy));
}
private ScanInfo preCompactScannerOpen(CompactionRequestImpl request, ScanType scanType,
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/DefaultCompactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/DefaultCompactor.java
index 49d3e8e..afa2429 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/DefaultCompactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/DefaultCompactor.java
@@ -45,14 +45,14 @@ public class DefaultCompactor extends Compactor<StoreFileWriter> {
}
private final CellSinkFactory<StoreFileWriter> writerFactory =
- new CellSinkFactory<StoreFileWriter>() {
- @Override
- public StoreFileWriter createWriter(InternalScanner scanner,
- org.apache.hadoop.hbase.regionserver.compactions.Compactor.FileDetails fd,
- boolean shouldDropBehind, boolean major) throws IOException {
- return createTmpWriter(fd, shouldDropBehind, major);
- }
- };
+ new CellSinkFactory<StoreFileWriter>() {
+ @Override
+ public StoreFileWriter createWriter(InternalScanner scanner,
+ org.apache.hadoop.hbase.regionserver.compactions.Compactor.FileDetails fd,
+ boolean shouldDropBehind, boolean major) throws IOException {
+ return DefaultCompactor.this.createWriter(fd, shouldDropBehind, major);
+ }
+ };
/**
* Do a minor/major compaction on an explicit set of storefiles from a Store.
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
new file mode 100644
index 0000000..d4c9a86
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
@@ -0,0 +1,61 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.regionserver.StoreContext;
+import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * The default implementation for store file tracker, where we do not persist the store file list,
+ * and use listing when loading store files.
+ */
+@InterfaceAudience.Private
+class DefaultStoreFileTracker extends StoreFileTrackerBase {
+
+ public DefaultStoreFileTracker(Configuration conf, TableName tableName, boolean isPrimaryReplica,
+ StoreContext ctx) {
+ super(conf, tableName, isPrimaryReplica, ctx);
+ }
+
+ @Override
+ public List<StoreFileInfo> load() throws IOException {
+ return ctx.getRegionFileSystem().getStoreFiles(ctx.getFamily().getNameAsString());
+ }
+
+ @Override
+ public boolean requireWritingToTmpDirFirst() {
+ return true;
+ }
+
+ @Override
+ protected void doAddNewStoreFiles(Collection<StoreFileInfo> newFiles) throws IOException {
+ // NOOP
+ }
+
+ @Override
+ protected void doAddCompactionResults(Collection<StoreFileInfo> compactedFiles,
+ Collection<StoreFileInfo> newFiles) throws IOException {
+ // NOOP
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java
new file mode 100644
index 0000000..aadedc8
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java
@@ -0,0 +1,75 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import org.apache.hadoop.hbase.regionserver.CreateStoreFileWriterParams;
+import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * An interface to define how we track the store files for a give store.
+ * <p/>
+ * In the old time, we will write store to a tmp directory first, and then rename it to the actual
+ * data file. And once a store file is under data directory, we will consider it as 'committed'. And
+ * we need to do listing when loading store files.
+ * <p/>
+ * When cloud age is coming, now we want to store the store files on object storage, where rename
+ * and list are not as cheap as on HDFS, especially rename. Although introducing a metadata
+ * management layer for object storage could solve the problem, but we still want HBase to run on
+ * pure object storage, so here we introduce this interface to abstract how we track the store
+ * files. For the old implementation, we just persist nothing here, and do listing to load store
+ * files. When running on object storage, we could persist the store file list in a system region,
+ * or in a file on the object storage, to make it possible to write directly into the data directory
+ * to avoid renaming, and also avoid listing when loading store files.
+ * <p/>
+ * The implementation requires to be thread safe as flush and compaction may occur as the same time,
+ * and we could also do multiple compactions at the same time. As the implementation may choose to
+ * persist the store file list to external storage, which could be slow, it is the duty for the
+ * callers to not call it inside a lock which may block normal read/write requests.
+ */
+@InterfaceAudience.Private
+public interface StoreFileTracker {
+
+ /**
+ * Load the store files list when opening a region.
+ */
+ List<StoreFileInfo> load() throws IOException;
+
+ /**
+ * Add new store files.
+ * <p/>
+ * Used for flush and bulk load.
+ */
+ void add(Collection<StoreFileInfo> newFiles) throws IOException;
+
+ /**
+ * Add new store files and remove compacted store files after compaction.
+ */
+ void replace(Collection<StoreFileInfo> compactedFiles, Collection<StoreFileInfo> newFiles)
+ throws IOException;
+
+ /**
+ * Create a writer for writing new store files.
+ * @return Writer for a new StoreFile
+ */
+ StoreFileWriter createWriter(CreateStoreFileWriterParams params) throws IOException;
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
new file mode 100644
index 0000000..2451f45
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
@@ -0,0 +1,178 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import java.io.IOException;
+import java.util.Collection;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.io.compress.Compression;
+import org.apache.hadoop.hbase.io.crypto.Encryption;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.io.hfile.HFile;
+import org.apache.hadoop.hbase.io.hfile.HFileContext;
+import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
+import org.apache.hadoop.hbase.regionserver.CreateStoreFileWriterParams;
+import org.apache.hadoop.hbase.regionserver.StoreContext;
+import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
+import org.apache.hadoop.hbase.regionserver.StoreUtils;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base class for all store file tracker.
+ * <p/>
+ * Mainly used to place the common logic to skip persistent for secondary replicas.
+ */
+@InterfaceAudience.Private
+abstract class StoreFileTrackerBase implements StoreFileTracker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StoreFileTrackerBase.class);
+
+ protected final Configuration conf;
+
+ protected final TableName tableName;
+
+ protected final boolean isPrimaryReplica;
+
+ protected final StoreContext ctx;
+
+ private volatile boolean cacheOnWriteLogged;
+
+ protected StoreFileTrackerBase(Configuration conf, TableName tableName, boolean isPrimaryReplica,
+ StoreContext ctx) {
+ this.conf = conf;
+ this.tableName = tableName;
+ this.isPrimaryReplica = isPrimaryReplica;
+ this.ctx = ctx;
+ }
+
+ @Override
+ public final void add(Collection<StoreFileInfo> newFiles) throws IOException {
+ if (isPrimaryReplica) {
+ doAddNewStoreFiles(newFiles);
+ }
+ }
+
+ @Override
+ public final void replace(Collection<StoreFileInfo> compactedFiles,
+ Collection<StoreFileInfo> newFiles) throws IOException {
+ if (isPrimaryReplica) {
+ doAddCompactionResults(compactedFiles, newFiles);
+ }
+ }
+
+ private HFileContext createFileContext(Compression.Algorithm compression,
+ boolean includeMVCCReadpoint, boolean includesTag, Encryption.Context encryptionContext) {
+ if (compression == null) {
+ compression = HFile.DEFAULT_COMPRESSION_ALGORITHM;
+ }
+ ColumnFamilyDescriptor family = ctx.getFamily();
+ HFileContext hFileContext = new HFileContextBuilder().withIncludesMvcc(includeMVCCReadpoint)
+ .withIncludesTags(includesTag).withCompression(compression)
+ .withCompressTags(family.isCompressTags()).withChecksumType(StoreUtils.getChecksumType(conf))
+ .withBytesPerCheckSum(StoreUtils.getBytesPerChecksum(conf))
+ .withBlockSize(family.getBlocksize()).withHBaseCheckSum(true)
+ .withDataBlockEncoding(family.getDataBlockEncoding()).withEncryptionContext(encryptionContext)
+ .withCreateTime(EnvironmentEdgeManager.currentTime()).withColumnFamily(family.getName())
+ .withTableName(tableName.getName()).withCellComparator(ctx.getComparator()).build();
+ return hFileContext;
+ }
+
+ @Override
+ public final StoreFileWriter createWriter(CreateStoreFileWriterParams params)
+ throws IOException {
+ if (!isPrimaryReplica) {
+ throw new IllegalStateException("Should not call create writer on secondary replicas");
+ }
+ // creating new cache config for each new writer
+ final CacheConfig cacheConf = ctx.getCacheConf();
+ final CacheConfig writerCacheConf = new CacheConfig(cacheConf);
+ long totalCompactedFilesSize = params.totalCompactedFilesSize();
+ if (params.isCompaction()) {
+ // Don't cache data on write on compactions, unless specifically configured to do so
+ // Cache only when total file size remains lower than configured threshold
+ final boolean cacheCompactedBlocksOnWrite = cacheConf.shouldCacheCompactedBlocksOnWrite();
+ // if data blocks are to be cached on write
+ // during compaction, we should forcefully
+ // cache index and bloom blocks as well
+ if (cacheCompactedBlocksOnWrite &&
+ totalCompactedFilesSize <= cacheConf.getCacheCompactedBlocksOnWriteThreshold()) {
+ writerCacheConf.enableCacheOnWrite();
+ if (!cacheOnWriteLogged) {
+ LOG.info("For {} , cacheCompactedBlocksOnWrite is true, hence enabled " +
+ "cacheOnWrite for Data blocks, Index blocks and Bloom filter blocks", this);
+ cacheOnWriteLogged = true;
+ }
+ } else {
+ writerCacheConf.setCacheDataOnWrite(false);
+ if (totalCompactedFilesSize > cacheConf.getCacheCompactedBlocksOnWriteThreshold()) {
+ // checking condition once again for logging
+ LOG.debug(
+ "For {}, setting cacheCompactedBlocksOnWrite as false as total size of compacted " +
+ "files - {}, is greater than cacheCompactedBlocksOnWriteThreshold - {}",
+ this, totalCompactedFilesSize, cacheConf.getCacheCompactedBlocksOnWriteThreshold());
+ }
+ }
+ } else {
+ final boolean shouldCacheDataOnWrite = cacheConf.shouldCacheDataOnWrite();
+ if (shouldCacheDataOnWrite) {
+ writerCacheConf.enableCacheOnWrite();
+ if (!cacheOnWriteLogged) {
+ LOG.info("For {} , cacheDataOnWrite is true, hence enabled cacheOnWrite for " +
+ "Index blocks and Bloom filter blocks", this);
+ cacheOnWriteLogged = true;
+ }
+ }
+ }
+ Encryption.Context encryptionContext = ctx.getEncryptionContext();
+ HFileContext hFileContext = createFileContext(params.compression(),
+ params.includeMVCCReadpoint(), params.includesTag(), encryptionContext);
+ Path outputDir;
+ if (requireWritingToTmpDirFirst()) {
+ outputDir =
+ new Path(ctx.getRegionFileSystem().getTempDir(), ctx.getFamily().getNameAsString());
+ } else {
+ throw new UnsupportedOperationException("not supported yet");
+ }
+ StoreFileWriter.Builder builder =
+ new StoreFileWriter.Builder(conf, writerCacheConf, ctx.getRegionFileSystem().getFileSystem())
+ .withOutputDir(outputDir).withBloomType(ctx.getBloomFilterType())
+ .withMaxKeyCount(params.maxKeyCount()).withFavoredNodes(ctx.getFavoredNodes())
+ .withFileContext(hFileContext).withShouldDropCacheBehind(params.shouldDropBehind())
+ .withCompactedFilesSupplier(ctx.getCompactedFilesSupplier())
+ .withFileStoragePolicy(params.fileStoragePolicy());
+ return builder.build();
+ }
+
+ /**
+ * Whether the implementation of this tracker requires you to write to temp directory first, i.e,
+ * does not allow broken store files under the actual data directory.
+ */
+ protected abstract boolean requireWritingToTmpDirFirst();
+
+ protected abstract void doAddNewStoreFiles(Collection<StoreFileInfo> newFiles) throws IOException;
+
+ protected abstract void doAddCompactionResults(Collection<StoreFileInfo> compactedFiles,
+ Collection<StoreFileInfo> newFiles) throws IOException;
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
new file mode 100644
index 0000000..4f7231b
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
@@ -0,0 +1,35 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.regionserver.StoreContext;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * Factory method for creating store file tracker.
+ */
+@InterfaceAudience.Private
+public final class StoreFileTrackerFactory {
+
+ public static StoreFileTracker create(Configuration conf, TableName tableName,
+ boolean isPrimaryReplica, StoreContext ctx) {
+ return new DefaultStoreFileTracker(conf, tableName, isPrimaryReplica, ctx);
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java
index 22ec6cb..291b909 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/compaction/MajorCompactionRequest.java
@@ -105,7 +105,6 @@ class MajorCompactionRequest {
boolean shouldCFBeCompacted(HRegionFileSystem fileSystem, String family, long ts)
throws IOException {
-
// do we have any store files?
Collection<StoreFileInfo> storeFiles = fileSystem.getStoreFiles(family);
if (storeFiles == null) {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/TestIOFencing.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/TestIOFencing.java
index 3c2bc3f..9314d7e 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/TestIOFencing.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/TestIOFencing.java
@@ -211,11 +211,13 @@ public class TestIOFencing {
@Override
protected void refreshStoreSizeAndTotalBytes() throws IOException {
- try {
- r.compactionsWaiting.countDown();
- r.compactionsBlocked.await();
- } catch (InterruptedException ex) {
- throw new IOException(ex);
+ if (r != null) {
+ try {
+ r.compactionsWaiting.countDown();
+ r.compactionsBlocked.await();
+ } catch (InterruptedException ex) {
+ throw new IOException(ex);
+ }
}
super.refreshStoreSizeAndTotalBytes();
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java
index cd83dc8..0a6b37b 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCacheOnWriteInSchema.java
@@ -215,8 +215,10 @@ public class TestCacheOnWriteInSchema {
@Test
public void testCacheOnWriteInSchema() throws IOException {
// Write some random data into the store
- StoreFileWriter writer = store.createWriterInTmp(Integer.MAX_VALUE,
- HFile.DEFAULT_COMPRESSION_ALGORITHM, false, true, false, false);
+ StoreFileWriter writer = store.getStoreEngine()
+ .createWriter(CreateStoreFileWriterParams.create().maxKeyCount(Integer.MAX_VALUE)
+ .compression(HFile.DEFAULT_COMPRESSION_ALGORITHM).isCompaction(false)
+ .includeMVCCReadpoint(true).includesTag(false).shouldDropBehind(false));
writeStoreFile(writer);
writer.close();
// Verify the block types of interest were cached on write
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultStoreEngine.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultStoreEngine.java
index e832c47..3784876 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultStoreEngine.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultStoreEngine.java
@@ -65,9 +65,12 @@ public class TestDefaultStoreEngine {
DummyCompactionPolicy.class.getName());
conf.set(DefaultStoreEngine.DEFAULT_STORE_FLUSHER_CLASS_KEY,
DummyStoreFlusher.class.getName());
+ HRegion mockRegion = Mockito.mock(HRegion.class);
HStore mockStore = Mockito.mock(HStore.class);
Mockito.when(mockStore.getRegionInfo()).thenReturn(RegionInfoBuilder.FIRST_META_REGIONINFO);
- StoreEngine<?, ?, ?, ?> se = StoreEngine.create(mockStore, conf, CellComparatorImpl.COMPARATOR);
+ Mockito.when(mockStore.getHRegion()).thenReturn(mockRegion);
+ StoreEngine<?, ?, ?, ?> se =
+ StoreEngine.create(mockStore, conf, CellComparatorImpl.COMPARATOR);
Assert.assertTrue(se instanceof DefaultStoreEngine);
Assert.assertTrue(se.getCompactionPolicy() instanceof DummyCompactionPolicy);
Assert.assertTrue(se.getStoreFlusher() instanceof DummyStoreFlusher);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
index a5584ff..7fb5768 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
@@ -5744,7 +5744,7 @@ public class TestHRegion {
Collection<HStoreFile> storeFiles = primaryRegion.getStore(families[0]).getStorefiles();
primaryRegion.getRegionFileSystem().removeStoreFiles(Bytes.toString(families[0]), storeFiles);
Collection<StoreFileInfo> storeFileInfos = primaryRegion.getRegionFileSystem()
- .getStoreFiles(families[0]);
+ .getStoreFiles(Bytes.toString(families[0]));
Assert.assertTrue(storeFileInfos == null || storeFileInfos.isEmpty());
verifyData(secondaryRegion, 0, 1000, cq, families);
@@ -7648,7 +7648,7 @@ public class TestHRegion {
getCacheConfig() != null? getCacheConfig().shouldEvictOnClose(): true;
for (Path newFile : newFiles) {
// Create storefile around what we wrote with a reader on it.
- HStoreFile sf = createStoreFileAndReader(newFile);
+ HStoreFile sf = storeEngine.createStoreFileAndReader(newFile);
sf.closeStoreFile(evictOnClose);
sfs.add(sf);
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java
index b64f521..8e4c4a3 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java
@@ -310,7 +310,7 @@ public class TestHStore {
/**
* Verify that compression and data block encoding are respected by the
- * Store.createWriterInTmp() method, used on store flush.
+ * createWriter method, used on store flush.
*/
@Test
public void testCreateWriter() throws Exception {
@@ -322,9 +322,11 @@ public class TestHStore {
.build();
init(name.getMethodName(), conf, hcd);
- // Test createWriterInTmp()
- StoreFileWriter writer =
- store.createWriterInTmp(4, hcd.getCompressionType(), false, true, false, false);
+ // Test createWriter
+ StoreFileWriter writer = store.getStoreEngine()
+ .createWriter(CreateStoreFileWriterParams.create().maxKeyCount(4)
+ .compression(hcd.getCompressionType()).isCompaction(false).includeMVCCReadpoint(true)
+ .includesTag(false).shouldDropBehind(false));
Path path = writer.getPath();
writer.append(new KeyValue(row, family, qf1, Bytes.toBytes(1)));
writer.append(new KeyValue(row, family, qf2, Bytes.toBytes(2)));
@@ -1021,19 +1023,19 @@ public class TestHStore {
// add one more file
addStoreFile();
- HStore spiedStore = spy(store);
+ StoreEngine<?, ?, ?, ?> spiedStoreEngine = spy(store.getStoreEngine());
// call first time after files changed
- spiedStore.refreshStoreFiles();
+ spiedStoreEngine.refreshStoreFiles();
assertEquals(2, this.store.getStorefilesCount());
- verify(spiedStore, times(1)).replaceStoreFiles(any(), any());
+ verify(spiedStoreEngine, times(1)).replaceStoreFiles(any(), any());
// call second time
- spiedStore.refreshStoreFiles();
+ spiedStoreEngine.refreshStoreFiles();
// ensure that replaceStoreFiles is not called, i.e, the times does not change, if files are not
// refreshed,
- verify(spiedStore, times(1)).replaceStoreFiles(any(), any());
+ verify(spiedStoreEngine, times(1)).replaceStoreFiles(any(), any());
}
private long countMemStoreScanner(StoreScanner scanner) {
@@ -1644,7 +1646,7 @@ public class TestHStore {
// Do compaction
MyThread thread = new MyThread(storeScanner);
thread.start();
- store.replaceStoreFiles(actualStorefiles, actualStorefiles1);
+ store.replaceStoreFiles(actualStorefiles, actualStorefiles1, false);
thread.join();
KeyValueHeap heap2 = thread.getHeap();
assertFalse(heap.equals(heap2));
@@ -1710,8 +1712,10 @@ public class TestHStore {
@Test
public void testHFileContextSetWithCFAndTable() throws Exception {
init(this.name.getMethodName());
- StoreFileWriter writer = store.createWriterInTmp(10000L,
- Compression.Algorithm.NONE, false, true, false, true);
+ StoreFileWriter writer = store.getStoreEngine()
+ .createWriter(CreateStoreFileWriterParams.create().maxKeyCount(10000L)
+ .compression(Compression.Algorithm.NONE).isCompaction(true).includeMVCCReadpoint(true)
+ .includesTag(false).shouldDropBehind(true));
HFileContext hFileContext = writer.getHFileWriter().getFileContext();
assertArrayEquals(family, hFileContext.getColumnFamily());
assertArrayEquals(table, hFileContext.getTableName());
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionMergeTransactionOnCluster.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionMergeTransactionOnCluster.java
index 60f65be..1dbf37a 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionMergeTransactionOnCluster.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionMergeTransactionOnCluster.java
@@ -245,7 +245,7 @@ public class TestRegionMergeTransactionOnCluster {
TEST_UTIL.getConfiguration(), fs, tabledir, mergedRegionInfo);
int count = 0;
for(ColumnFamilyDescriptor colFamily : columnFamilies) {
- count += hrfs.getStoreFiles(colFamily.getName()).size();
+ count += hrfs.getStoreFiles(colFamily.getNameAsString()).size();
}
ADMIN.compactRegion(mergedRegionInfo.getRegionName());
// clean up the merged region store files
@@ -254,7 +254,7 @@ public class TestRegionMergeTransactionOnCluster {
int newcount = 0;
while (EnvironmentEdgeManager.currentTime() < timeout) {
for(ColumnFamilyDescriptor colFamily : columnFamilies) {
- newcount += hrfs.getStoreFiles(colFamily.getName()).size();
+ newcount += hrfs.getStoreFiles(colFamily.getNameAsString()).size();
}
if(newcount > count) {
break;
@@ -273,7 +273,7 @@ public class TestRegionMergeTransactionOnCluster {
while (EnvironmentEdgeManager.currentTime() < timeout) {
int newcount1 = 0;
for(ColumnFamilyDescriptor colFamily : columnFamilies) {
- newcount1 += hrfs.getStoreFiles(colFamily.getName()).size();
+ newcount1 += hrfs.getStoreFiles(colFamily.getNameAsString()).size();
}
if(newcount1 <= 1) {
break;
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java
index 9141327..2fab050 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java
@@ -26,7 +26,6 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
@@ -98,7 +97,7 @@ public class TestStoreFileRefresherChore {
}
@Override
- public Collection<StoreFileInfo> getStoreFiles(String familyName) throws IOException {
+ public List<StoreFileInfo> getStoreFiles(String familyName) throws IOException {
if (fail) {
throw new IOException("simulating FS failure");
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScannerClosure.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScannerClosure.java
index 4cb023f..121f830 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScannerClosure.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreScannerClosure.java
@@ -29,7 +29,6 @@ import java.util.NavigableSet;
import java.util.Random;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -128,13 +127,12 @@ public class TestStoreScannerClosure {
p.addColumn(fam, Bytes.toBytes("q1"), Bytes.toBytes("val"));
region.put(p);
HStore store = region.getStore(fam);
- ReentrantReadWriteLock lock = store.lock;
// use the lock to manually get a new memstore scanner. this is what
// HStore#notifyChangedReadersObservers does under the lock.(lock is not needed here
//since it is just a testcase).
- lock.readLock().lock();
+ store.getStoreEngine().readLock();
final List<KeyValueScanner> memScanners = store.memstore.getScanners(Long.MAX_VALUE);
- lock.readLock().unlock();
+ store.getStoreEngine().readUnlock();
Thread closeThread = new Thread() {
public void run() {
// close should be completed
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStripeStoreEngine.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStripeStoreEngine.java
index f5330f6..eb0b1c1 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStripeStoreEngine.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStripeStoreEngine.java
@@ -118,8 +118,10 @@ public class TestStripeStoreEngine {
}
private static TestStoreEngine createEngine(Configuration conf) throws Exception {
+ HRegion region = mock(HRegion.class);
HStore store = mock(HStore.class);
when(store.getRegionInfo()).thenReturn(RegionInfoBuilder.FIRST_META_REGIONINFO);
+ when(store.getHRegion()).thenReturn(region);
CellComparatorImpl kvComparator = mock(CellComparatorImpl.class);
return (TestStoreEngine) StoreEngine.create(store, conf, kvComparator);
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestDateTieredCompactor.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestDateTieredCompactor.java
index d7b7ba7..0ea82c5 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestDateTieredCompactor.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestDateTieredCompactor.java
@@ -22,9 +22,6 @@ import static org.apache.hadoop.hbase.regionserver.compactions.TestCompactor.cre
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -45,11 +42,13 @@ import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.RegionInfoBuilder;
+import org.apache.hadoop.hbase.regionserver.CreateStoreFileWriterParams;
import org.apache.hadoop.hbase.regionserver.HStore;
import org.apache.hadoop.hbase.regionserver.HStoreFile;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.ScanInfo;
import org.apache.hadoop.hbase.regionserver.ScanType;
+import org.apache.hadoop.hbase.regionserver.StoreEngine;
import org.apache.hadoop.hbase.regionserver.StoreFileScanner;
import org.apache.hadoop.hbase.regionserver.StoreUtils;
import org.apache.hadoop.hbase.regionserver.compactions.TestCompactor.Scanner;
@@ -110,10 +109,9 @@ public class TestDateTieredCompactor {
when(store.areWritesEnabled()).thenReturn(true);
when(store.getFileSystem()).thenReturn(mock(FileSystem.class));
when(store.getRegionInfo()).thenReturn(RegionInfoBuilder.newBuilder(TABLE_NAME).build());
- when(store.createWriterInTmp(anyLong(), any(), anyBoolean(),
- anyBoolean(), anyBoolean(), anyBoolean())).thenAnswer(writers);
- when(store.createWriterInTmp(anyLong(), any(), anyBoolean(),
- anyBoolean(), anyBoolean(), anyBoolean(), anyLong(), anyString())).thenAnswer(writers);
+ StoreEngine storeEngine = mock(StoreEngine.class);
+ when(storeEngine.createWriter(any(CreateStoreFileWriterParams.class))).thenAnswer(writers);
+ when(store.getStoreEngine()).thenReturn(storeEngine);
when(store.getComparator()).thenReturn(CellComparatorImpl.COMPARATOR);
OptionalLong maxSequenceId = StoreUtils.getMaxSequenceIdInList(storefiles);
when(store.getMaxSequenceId()).thenReturn(maxSequenceId);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestStripeCompactionPolicy.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestStripeCompactionPolicy.java
index 2ba15d1..bdab20e 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestStripeCompactionPolicy.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestStripeCompactionPolicy.java
@@ -30,7 +30,6 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
@@ -60,6 +59,7 @@ import org.apache.hadoop.hbase.client.RegionInfoBuilder;
import org.apache.hadoop.hbase.io.TimeRange;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.regionserver.BloomType;
+import org.apache.hadoop.hbase.regionserver.CreateStoreFileWriterParams;
import org.apache.hadoop.hbase.regionserver.HStore;
import org.apache.hadoop.hbase.regionserver.HStoreFile;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
@@ -67,6 +67,7 @@ import org.apache.hadoop.hbase.regionserver.ScanInfo;
import org.apache.hadoop.hbase.regionserver.ScanType;
import org.apache.hadoop.hbase.regionserver.ScannerContext;
import org.apache.hadoop.hbase.regionserver.StoreConfigInformation;
+import org.apache.hadoop.hbase.regionserver.StoreEngine;
import org.apache.hadoop.hbase.regionserver.StoreFileReader;
import org.apache.hadoop.hbase.regionserver.StoreFileScanner;
import org.apache.hadoop.hbase.regionserver.StripeMultiFileWriter;
@@ -864,12 +865,9 @@ public class TestStripeCompactionPolicy {
when(info.getRegionNameAsString()).thenReturn("testRegion");
when(store.getColumnFamilyDescriptor()).thenReturn(familyDescriptor);
when(store.getRegionInfo()).thenReturn(info);
- when(
- store.createWriterInTmp(anyLong(), any(), anyBoolean(),
- anyBoolean(), anyBoolean(), anyBoolean())).thenAnswer(writers);
- when(
- store.createWriterInTmp(anyLong(), any(), anyBoolean(),
- anyBoolean(), anyBoolean(), anyBoolean(), anyLong(), anyString())).thenAnswer(writers);
+ StoreEngine storeEngine = mock(StoreEngine.class);
+ when(storeEngine.createWriter(any(CreateStoreFileWriterParams.class))).thenAnswer(writers);
+ when(store.getStoreEngine()).thenReturn(storeEngine);
Configuration conf = HBaseConfiguration.create();
conf.setBoolean("hbase.regionserver.compaction.private.readers", usePrivateReaders);
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestStripeCompactor.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestStripeCompactor.java
index e49174e..ae59c74 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestStripeCompactor.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/compactions/TestStripeCompactor.java
@@ -21,9 +21,6 @@ import static org.apache.hadoop.hbase.regionserver.StripeStoreFileManager.OPEN_K
import static org.apache.hadoop.hbase.regionserver.compactions.TestCompactor.createDummyRequest;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -43,10 +40,12 @@ import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.RegionInfoBuilder;
+import org.apache.hadoop.hbase.regionserver.CreateStoreFileWriterParams;
import org.apache.hadoop.hbase.regionserver.HStore;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.ScanInfo;
import org.apache.hadoop.hbase.regionserver.ScanType;
+import org.apache.hadoop.hbase.regionserver.StoreEngine;
import org.apache.hadoop.hbase.regionserver.StoreFileScanner;
import org.apache.hadoop.hbase.regionserver.compactions.TestCompactor.Scanner;
import org.apache.hadoop.hbase.regionserver.compactions.TestCompactor.StoreFileWritersCapture;
@@ -209,10 +208,9 @@ public class TestStripeCompactor {
when(store.areWritesEnabled()).thenReturn(true);
when(store.getFileSystem()).thenReturn(mock(FileSystem.class));
when(store.getRegionInfo()).thenReturn(RegionInfoBuilder.newBuilder(TABLE_NAME).build());
- when(store.createWriterInTmp(anyLong(), any(), anyBoolean(),
- anyBoolean(), anyBoolean(), anyBoolean())).thenAnswer(writers);
- when(store.createWriterInTmp(anyLong(), any(), anyBoolean(),
- anyBoolean(), anyBoolean(), anyBoolean(), anyLong(), anyString())).thenAnswer(writers);
+ StoreEngine storeEngine = mock(StoreEngine.class);
+ when(storeEngine.createWriter(any(CreateStoreFileWriterParams.class))).thenAnswer(writers);
+ when(store.getStoreEngine()).thenReturn(storeEngine);
when(store.getComparator()).thenReturn(CellComparatorImpl.COMPARATOR);
return new StripeCompactor(conf, store) {
[hbase] 03/05: HBASE-26079 Use StoreFileTracker when splitting and
merging (#3617)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-26067
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit bd924ab8d97977cbcc35f18c52baf786f0b7b0d2
Author: Wellington Ramos Chevreuil <wc...@apache.org>
AuthorDate: Wed Sep 8 10:31:49 2021 +0100
HBASE-26079 Use StoreFileTracker when splitting and merging (#3617)
Signed-off-by: Duo Zhang <zh...@apache.org>
---
.../assignment/MergeTableRegionsProcedure.java | 23 +-
.../assignment/SplitTableRegionProcedure.java | 41 ++--
.../hbase/regionserver/HRegionFileSystem.java | 42 +++-
.../storefiletracker/DefaultStoreFileTracker.java | 4 +-
.../storefiletracker/StoreFileTracker.java | 1 -
.../storefiletracker/StoreFileTrackerFactory.java | 33 ++-
.../hbase/regionserver/TestDefaultStoreEngine.java | 1 +
.../regionserver/TestDirectStoreSplitsMerges.java | 32 ++-
.../hadoop/hbase/regionserver/TestHStoreFile.java | 19 +-
.../regionserver/TestMergesSplitsAddToTracker.java | 262 +++++++++++++++++++++
.../hbase/regionserver/TestStripeStoreEngine.java | 1 +
.../storefiletracker/TestStoreFileTracker.java} | 42 ++--
12 files changed, 436 insertions(+), 65 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/MergeTableRegionsProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/MergeTableRegionsProcedure.java
index fb57cb9..4b25a28 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/MergeTableRegionsProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/MergeTableRegionsProcedure.java
@@ -24,6 +24,8 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.stream.Stream;
+
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.MetaMutationAnnotation;
@@ -53,6 +55,8 @@ import org.apache.hadoop.hbase.quotas.QuotaExceededException;
import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
import org.apache.hadoop.hbase.regionserver.HStoreFile;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.wal.WALSplitUtil;
@@ -584,40 +588,47 @@ public class MergeTableRegionsProcedure
final MasterFileSystem mfs = env.getMasterServices().getMasterFileSystem();
final Path tableDir = CommonFSUtils.getTableDir(mfs.getRootDir(), regionsToMerge[0].getTable());
final FileSystem fs = mfs.getFileSystem();
-
+ List<Path> mergedFiles = new ArrayList<>();
HRegionFileSystem mergeRegionFs = HRegionFileSystem.createRegionOnFileSystem(
env.getMasterConfiguration(), fs, tableDir, mergedRegion);
for (RegionInfo ri: this.regionsToMerge) {
HRegionFileSystem regionFs = HRegionFileSystem.openRegionFromFileSystem(
env.getMasterConfiguration(), fs, tableDir, ri, false);
- mergeStoreFiles(env, regionFs, mergeRegionFs, mergedRegion);
+ mergedFiles.addAll(mergeStoreFiles(env, regionFs, mergeRegionFs, mergedRegion));
}
assert mergeRegionFs != null;
- mergeRegionFs.commitMergedRegion();
+ mergeRegionFs.commitMergedRegion(mergedFiles, env);
// Prepare to create merged regions
env.getAssignmentManager().getRegionStates().
getOrCreateRegionStateNode(mergedRegion).setState(State.MERGING_NEW);
}
- private void mergeStoreFiles(MasterProcedureEnv env, HRegionFileSystem regionFs,
+ private List<Path> mergeStoreFiles(MasterProcedureEnv env, HRegionFileSystem regionFs,
HRegionFileSystem mergeRegionFs, RegionInfo mergedRegion) throws IOException {
final TableDescriptor htd = env.getMasterServices().getTableDescriptors()
.get(mergedRegion.getTable());
+ List<Path> mergedFiles = new ArrayList<>();
for (ColumnFamilyDescriptor hcd : htd.getColumnFamilies()) {
String family = hcd.getNameAsString();
- final Collection<StoreFileInfo> storeFiles = regionFs.getStoreFiles(family);
+ Configuration trackerConfig =
+ StoreFileTrackerFactory.mergeConfigurations(env.getMasterConfiguration(), htd, hcd);
+ StoreFileTracker tracker = StoreFileTrackerFactory.create(trackerConfig, true,
+ family, regionFs);
+ final Collection<StoreFileInfo> storeFiles = tracker.load();
if (storeFiles != null && storeFiles.size() > 0) {
for (StoreFileInfo storeFileInfo : storeFiles) {
// Create reference file(s) to parent region file here in mergedDir.
// As this procedure is running on master, use CacheConfig.DISABLED means
// don't cache any block.
- mergeRegionFs.mergeStoreFile(regionFs.getRegionInfo(), family,
+ Path refFile = mergeRegionFs.mergeStoreFile(regionFs.getRegionInfo(), family,
new HStoreFile(storeFileInfo, hcd.getBloomFilterType(), CacheConfig.DISABLED));
+ mergedFiles.add(refFile);
}
}
}
+ return mergedFiles;
}
/**
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java
index 0a15e36..4d53df4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java
@@ -64,6 +64,8 @@ import org.apache.hadoop.hbase.regionserver.HStoreFile;
import org.apache.hadoop.hbase.regionserver.RegionSplitPolicy;
import org.apache.hadoop.hbase.regionserver.RegionSplitRestriction;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
@@ -618,21 +620,20 @@ public class SplitTableRegionProcedure
final FileSystem fs = mfs.getFileSystem();
HRegionFileSystem regionFs = HRegionFileSystem.openRegionFromFileSystem(
env.getMasterConfiguration(), fs, tabledir, getParentRegion(), false);
-
regionFs.createSplitsDir(daughterOneRI, daughterTwoRI);
- Pair<Integer, Integer> expectedReferences = splitStoreFiles(env, regionFs);
+ Pair<List<Path>, List<Path>> expectedReferences = splitStoreFiles(env, regionFs);
- assertReferenceFileCount(fs, expectedReferences.getFirst(),
+ assertReferenceFileCount(fs, expectedReferences.getFirst().size(),
regionFs.getSplitsDir(daughterOneRI));
- regionFs.commitDaughterRegion(daughterOneRI);
- assertReferenceFileCount(fs, expectedReferences.getFirst(),
+ regionFs.commitDaughterRegion(daughterOneRI, expectedReferences.getFirst(), env);
+ assertReferenceFileCount(fs, expectedReferences.getFirst().size(),
new Path(tabledir, daughterOneRI.getEncodedName()));
- assertReferenceFileCount(fs, expectedReferences.getSecond(),
+ assertReferenceFileCount(fs, expectedReferences.getSecond().size(),
regionFs.getSplitsDir(daughterTwoRI));
- regionFs.commitDaughterRegion(daughterTwoRI);
- assertReferenceFileCount(fs, expectedReferences.getSecond(),
+ regionFs.commitDaughterRegion(daughterTwoRI, expectedReferences.getSecond(), env);
+ assertReferenceFileCount(fs, expectedReferences.getSecond().size(),
new Path(tabledir, daughterTwoRI.getEncodedName()));
}
@@ -649,7 +650,7 @@ public class SplitTableRegionProcedure
* Create Split directory
* @param env MasterProcedureEnv
*/
- private Pair<Integer, Integer> splitStoreFiles(final MasterProcedureEnv env,
+ private Pair<List<Path>, List<Path>> splitStoreFiles(final MasterProcedureEnv env,
final HRegionFileSystem regionFs) throws IOException {
final Configuration conf = env.getMasterConfiguration();
TableDescriptor htd = env.getMasterServices().getTableDescriptors().get(getTableName());
@@ -665,7 +666,11 @@ public class SplitTableRegionProcedure
new HashMap<String, Collection<StoreFileInfo>>(htd.getColumnFamilyCount());
for (ColumnFamilyDescriptor cfd : htd.getColumnFamilies()) {
String family = cfd.getNameAsString();
- Collection<StoreFileInfo> sfis = regionFs.getStoreFiles(family);
+ Configuration trackerConfig = StoreFileTrackerFactory.
+ mergeConfigurations(env.getMasterConfiguration(), htd, htd.getColumnFamily(cfd.getName()));
+ StoreFileTracker tracker = StoreFileTrackerFactory.create(trackerConfig, true,
+ family, regionFs);
+ Collection<StoreFileInfo> sfis = tracker.load();
if (sfis == null) {
continue;
}
@@ -691,7 +696,7 @@ public class SplitTableRegionProcedure
}
if (nbFiles == 0) {
// no file needs to be splitted.
- return new Pair<Integer, Integer>(0, 0);
+ return new Pair<>(Collections.emptyList(), Collections.emptyList());
}
// Max #threads is the smaller of the number of storefiles or the default max determined above.
int maxThreads = Math.min(
@@ -744,14 +749,18 @@ public class SplitTableRegionProcedure
throw (InterruptedIOException) new InterruptedIOException().initCause(e);
}
- int daughterA = 0;
- int daughterB = 0;
+ List<Path> daughterA = new ArrayList<>();
+ List<Path> daughterB = new ArrayList<>();
// Look for any exception
for (Future<Pair<Path, Path>> future : futures) {
try {
Pair<Path, Path> p = future.get();
- daughterA += p.getFirst() != null ? 1 : 0;
- daughterB += p.getSecond() != null ? 1 : 0;
+ if(p.getFirst() != null){
+ daughterA.add(p.getFirst());
+ }
+ if(p.getSecond() != null){
+ daughterB.add(p.getSecond());
+ }
} catch (InterruptedException e) {
throw (InterruptedIOException) new InterruptedIOException().initCause(e);
} catch (ExecutionException e) {
@@ -764,7 +773,7 @@ public class SplitTableRegionProcedure
getParentRegion().getShortNameToLog() + " Daughter A: " + daughterA +
" storefiles, Daughter B: " + daughterB + " storefiles.");
}
- return new Pair<Integer, Integer>(daughterA, daughterB);
+ return new Pair<>(daughterA, daughterB);
}
private void assertReferenceFileCount(final FileSystem fs, final int expectedReferenceFileCount,
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
index 6ab9b0f..3b053a3 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
@@ -23,7 +23,9 @@ import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
@@ -45,6 +47,9 @@ import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.fs.HFileSystem;
import org.apache.hadoop.hbase.io.Reference;
+import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTracker;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.FSUtils;
@@ -592,19 +597,46 @@ public class HRegionFileSystem {
* @param regionInfo daughter {@link org.apache.hadoop.hbase.client.RegionInfo}
* @throws IOException
*/
- public Path commitDaughterRegion(final RegionInfo regionInfo)
- throws IOException {
+ public Path commitDaughterRegion(final RegionInfo regionInfo, List<Path> allRegionFiles,
+ MasterProcedureEnv env) throws IOException {
Path regionDir = this.getSplitsDir(regionInfo);
if (fs.exists(regionDir)) {
// Write HRI to a file in case we need to recover hbase:meta
Path regionInfoFile = new Path(regionDir, REGION_INFO_FILE);
byte[] regionInfoContent = getRegionInfoFileContent(regionInfo);
writeRegionInfoFileContent(conf, fs, regionInfoFile, regionInfoContent);
+ HRegionFileSystem regionFs = HRegionFileSystem.openRegionFromFileSystem(
+ env.getMasterConfiguration(), fs, getTableDir(), regionInfo, false);
+ insertRegionFilesIntoStoreTracker(allRegionFiles, env, regionFs);
}
-
return regionDir;
}
+ private void insertRegionFilesIntoStoreTracker(List<Path> allFiles, MasterProcedureEnv env,
+ HRegionFileSystem regionFs) throws IOException {
+ TableDescriptor tblDesc = env.getMasterServices().getTableDescriptors().
+ get(regionInfo.getTable());
+ //we need to map trackers per store
+ Map<String, StoreFileTracker> trackerMap = new HashMap<>();
+ //we need to map store files per store
+ Map<String, List<StoreFileInfo>> fileInfoMap = new HashMap<>();
+ for(Path file : allFiles) {
+ String familyName = file.getParent().getName();
+ trackerMap.computeIfAbsent(familyName, t -> {
+ Configuration config = StoreFileTrackerFactory.mergeConfigurations(conf, tblDesc,
+ tblDesc.getColumnFamily(Bytes.toBytes(familyName)));
+ return StoreFileTrackerFactory.
+ create(config, true, familyName, regionFs);
+ });
+ fileInfoMap.computeIfAbsent(familyName, l -> new ArrayList<>());
+ List<StoreFileInfo> infos = fileInfoMap.get(familyName);
+ infos.add(new StoreFileInfo(conf, fs, file, true));
+ }
+ for(Map.Entry<String, StoreFileTracker> entry : trackerMap.entrySet()) {
+ entry.getValue().add(fileInfoMap.get(entry.getKey()));
+ }
+ }
+
/**
* Creates region split daughter directories under the table dir. If the daughter regions already
* exist, for example, in the case of a recovery from a previous failed split procedure, this
@@ -755,13 +787,15 @@ public class HRegionFileSystem {
* Commit a merged region, making it ready for use.
* @throws IOException
*/
- public void commitMergedRegion() throws IOException {
+ public void commitMergedRegion(List<Path> allMergedFiles, MasterProcedureEnv env)
+ throws IOException {
Path regionDir = getMergesDir(regionInfoForFs);
if (regionDir != null && fs.exists(regionDir)) {
// Write HRI to a file in case we need to recover hbase:meta
Path regionInfoFile = new Path(regionDir, REGION_INFO_FILE);
byte[] regionInfoContent = getRegionInfoFileContent(regionInfo);
writeRegionInfoFileContent(conf, fs, regionInfoFile, regionInfoContent);
+ insertRegionFilesIntoStoreTracker(allMergedFiles, env, this);
}
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
index fa04481..22e0513 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.util.Collection;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
+
import org.apache.hadoop.hbase.regionserver.StoreContext;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
import org.apache.yetus.audience.InterfaceAudience;
@@ -32,8 +33,7 @@ import org.apache.yetus.audience.InterfaceAudience;
@InterfaceAudience.Private
class DefaultStoreFileTracker extends StoreFileTrackerBase {
- public DefaultStoreFileTracker(Configuration conf, boolean isPrimaryReplica,
- StoreContext ctx) {
+ public DefaultStoreFileTracker(Configuration conf, boolean isPrimaryReplica, StoreContext ctx) {
super(conf, isPrimaryReplica, ctx);
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java
index aadedc8..0a85abb 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTracker.java
@@ -48,7 +48,6 @@ import org.apache.yetus.audience.InterfaceAudience;
*/
@InterfaceAudience.Private
public interface StoreFileTracker {
-
/**
* Load the store files list when opening a region.
*/
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
index 6cdfaf4..c446d5a 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
@@ -18,22 +18,51 @@
package org.apache.hadoop.hbase.regionserver.storefiletracker;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.CompoundConfiguration;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
import org.apache.hadoop.hbase.regionserver.StoreContext;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Factory method for creating store file tracker.
*/
@InterfaceAudience.Private
public final class StoreFileTrackerFactory {
-
public static final String TRACK_IMPL = "hbase.store.file-tracker.impl";
+ private static final Logger LOG = LoggerFactory.getLogger(StoreFileTrackerFactory.class);
public static StoreFileTracker create(Configuration conf, boolean isPrimaryReplica,
- StoreContext ctx) {
+ StoreContext ctx) {
Class<? extends StoreFileTracker> tracker =
conf.getClass(TRACK_IMPL, DefaultStoreFileTracker.class, StoreFileTracker.class);
+ LOG.info("instantiating StoreFileTracker impl {}", tracker.getName());
return ReflectionUtils.newInstance(tracker, conf, isPrimaryReplica, ctx);
}
+
+ public static StoreFileTracker create(Configuration conf, boolean isPrimaryReplica, String family,
+ HRegionFileSystem regionFs) {
+ ColumnFamilyDescriptorBuilder fDescBuilder =
+ ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family));
+ StoreContext ctx = StoreContext.getBuilder().
+ withColumnFamilyDescriptor(fDescBuilder.build()).
+ withRegionFileSystem(regionFs).
+ build();
+ return StoreFileTrackerFactory.create(conf, isPrimaryReplica, ctx);
+ }
+
+ public static Configuration mergeConfigurations(Configuration global,
+ TableDescriptor table, ColumnFamilyDescriptor family) {
+ return new CompoundConfiguration()
+ .add(global)
+ .addBytesMap(table.getValues())
+ .addStringMap(family.getConfiguration())
+ .addBytesMap(family.getValues());
+ }
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultStoreEngine.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultStoreEngine.java
index 3784876..523f277 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultStoreEngine.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDefaultStoreEngine.java
@@ -67,6 +67,7 @@ public class TestDefaultStoreEngine {
DummyStoreFlusher.class.getName());
HRegion mockRegion = Mockito.mock(HRegion.class);
HStore mockStore = Mockito.mock(HStore.class);
+ mockStore.conf = conf;
Mockito.when(mockStore.getRegionInfo()).thenReturn(RegionInfoBuilder.FIRST_META_REGIONINFO);
Mockito.when(mockStore.getHRegion()).thenReturn(mockRegion);
StoreEngine<?, ?, ?, ?> se =
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDirectStoreSplitsMerges.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDirectStoreSplitsMerges.java
index bd24f1b..0eba8aa 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDirectStoreSplitsMerges.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestDirectStoreSplitsMerges.java
@@ -21,6 +21,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.fs.Path;
@@ -32,6 +33,7 @@ import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionInfoBuilder;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.master.assignment.SplitTableRegionProcedure;
+import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.procedure2.Procedure;
import org.apache.hadoop.hbase.testclassification.LargeTests;
import org.apache.hadoop.hbase.testclassification.RegionServerTests;
@@ -139,7 +141,9 @@ public class TestDirectStoreSplitsMerges {
setRegionId(region.getRegionInfo().getRegionId() +
EnvironmentEdgeManager.currentTime()).build();
Path splitDir = regionFS.getSplitsDir(daughterA);
- Path result = regionFS.commitDaughterRegion(daughterA);
+ MasterProcedureEnv env = TEST_UTIL.getMiniHBaseCluster().getMaster().
+ getMasterProcedureExecutor().getEnvironment();
+ Path result = regionFS.commitDaughterRegion(daughterA, new ArrayList<>(), env);
assertEquals(splitDir, result);
}
@@ -162,14 +166,18 @@ public class TestDirectStoreSplitsMerges {
Path splitDirA = regionFS.getSplitsDir(daughterA);
Path splitDirB = regionFS.getSplitsDir(daughterB);
HStoreFile file = (HStoreFile) region.getStore(FAMILY_NAME).getStorefiles().toArray()[0];
- regionFS
+ List<Path> filesA = new ArrayList<>();
+ filesA.add(regionFS
.splitStoreFile(daughterA, Bytes.toString(FAMILY_NAME), file,
- Bytes.toBytes("002"), false, region.getSplitPolicy());
- regionFS
+ Bytes.toBytes("002"), false, region.getSplitPolicy()));
+ List<Path> filesB = new ArrayList<>();
+ filesB.add(regionFS
.splitStoreFile(daughterB, Bytes.toString(FAMILY_NAME), file,
- Bytes.toBytes("002"), true, region.getSplitPolicy());
- Path resultA = regionFS.commitDaughterRegion(daughterA);
- Path resultB = regionFS.commitDaughterRegion(daughterB);
+ Bytes.toBytes("002"), true, region.getSplitPolicy()));
+ MasterProcedureEnv env = TEST_UTIL.getMiniHBaseCluster().getMaster().
+ getMasterProcedureExecutor().getEnvironment();
+ Path resultA = regionFS.commitDaughterRegion(daughterA, filesA, env);
+ Path resultB = regionFS.commitDaughterRegion(daughterB, filesB, env);
assertEquals(splitDirA, resultA);
assertEquals(splitDirB, resultB);
}
@@ -203,8 +211,11 @@ public class TestDirectStoreSplitsMerges {
mergeFileFromRegion(mergeRegionFs, first, file);
//merge file from second region
file = (HStoreFile) second.getStore(FAMILY_NAME).getStorefiles().toArray()[0];
- mergeFileFromRegion(mergeRegionFs, second, file);
- mergeRegionFs.commitMergedRegion();
+ List<Path> mergedFiles = new ArrayList<>();
+ mergedFiles.add(mergeFileFromRegion(mergeRegionFs, second, file));
+ MasterProcedureEnv env = TEST_UTIL.getMiniHBaseCluster().getMaster().
+ getMasterProcedureExecutor().getEnvironment();
+ mergeRegionFs.commitMergedRegion(mergedFiles, env);
}
private void waitForSplitProcComplete(int attempts, int waitTime) throws Exception {
@@ -223,11 +234,12 @@ public class TestDirectStoreSplitsMerges {
}
}
- private void mergeFileFromRegion(HRegionFileSystem regionFS, HRegion regionToMerge,
+ private Path mergeFileFromRegion(HRegionFileSystem regionFS, HRegion regionToMerge,
HStoreFile file) throws IOException {
Path mergedFile = regionFS.mergeStoreFile(regionToMerge.getRegionInfo(),
Bytes.toString(FAMILY_NAME), file);
validateResultingFile(regionToMerge.getRegionInfo().getEncodedName(), mergedFile);
+ return mergedFile;
}
private void validateResultingFile(String originalRegion, Path result){
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStoreFile.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStoreFile.java
index e6f0357..c7203a9 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStoreFile.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStoreFile.java
@@ -24,6 +24,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -49,12 +50,14 @@ import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.PrivateCellUtil;
+import org.apache.hadoop.hbase.TableDescriptors;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionInfoBuilder;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper;
import org.apache.hadoop.hbase.io.HFileLink;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
@@ -69,6 +72,8 @@ import org.apache.hadoop.hbase.io.hfile.HFileInfo;
import org.apache.hadoop.hbase.io.hfile.HFileScanner;
import org.apache.hadoop.hbase.io.hfile.ReaderContext;
import org.apache.hadoop.hbase.io.hfile.ReaderContextBuilder;
+import org.apache.hadoop.hbase.master.MasterServices;
+import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.testclassification.RegionServerTests;
import org.apache.hadoop.hbase.util.BloomFilterFactory;
@@ -1055,7 +1060,19 @@ public class TestHStoreFile {
if (null == path) {
return null;
}
- Path regionDir = regionFs.commitDaughterRegion(hri);
+ List<Path> splitFiles = new ArrayList<>();
+ splitFiles.add(path);
+ MasterProcedureEnv mockEnv = mock(MasterProcedureEnv.class);
+ MasterServices mockServices = mock(MasterServices.class);
+ when(mockEnv.getMasterServices()).thenReturn(mockServices);
+ when(mockEnv.getMasterConfiguration()).thenReturn(new Configuration());
+ TableDescriptors mockTblDescs = mock(TableDescriptors.class);
+ when(mockServices.getTableDescriptors()).thenReturn(mockTblDescs);
+ TableDescriptor mockTblDesc = mock(TableDescriptor.class);
+ when(mockTblDescs.get(any())).thenReturn(mockTblDesc);
+ ColumnFamilyDescriptor mockCfDesc = mock(ColumnFamilyDescriptor.class);
+ when(mockTblDesc.getColumnFamily(any())).thenReturn(mockCfDesc);
+ Path regionDir = regionFs.commitDaughterRegion(hri, splitFiles, mockEnv);
return new Path(new Path(regionDir, family), path.getName());
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMergesSplitsAddToTracker.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMergesSplitsAddToTracker.java
new file mode 100644
index 0000000..c6205cb
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMergesSplitsAddToTracker.java
@@ -0,0 +1,262 @@
+/**
+ * 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.hadoop.hbase.regionserver;
+
+import static org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory.
+ TRACK_IMPL;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.mutable.MutableBoolean;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.RegionInfoBuilder;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
+import org.apache.hadoop.hbase.regionserver.storefiletracker.TestStoreFileTracker;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.Pair;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+
+
+@Category({RegionServerTests.class, LargeTests.class})
+public class TestMergesSplitsAddToTracker {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestMergesSplitsAddToTracker.class);
+
+ private static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
+
+ public static final byte[] FAMILY_NAME = Bytes.toBytes("info");
+
+ @Rule
+ public TestName name = new TestName();
+
+ @BeforeClass
+ public static void setupClass() throws Exception {
+ TEST_UTIL.getConfiguration().set(TRACK_IMPL, TestStoreFileTracker.class.getName());
+ TEST_UTIL.startMiniCluster();
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ TEST_UTIL.shutdownMiniCluster();
+ }
+
+ @Before
+ public void setup(){
+ TestStoreFileTracker.trackedFiles = new HashMap<>();
+ }
+
+ @Test
+ public void testCommitDaughterRegion() throws Exception {
+ TableName table = TableName.valueOf(name.getMethodName());
+ TEST_UTIL.createTable(table, FAMILY_NAME);
+ //first put some data in order to have a store file created
+ putThreeRowsAndFlush(table);
+ HRegion region = TEST_UTIL.getHBaseCluster().getRegions(table).get(0);
+ HRegionFileSystem regionFS = region.getStores().get(0).getRegionFileSystem();
+ RegionInfo daughterA =
+ RegionInfoBuilder.newBuilder(table).setStartKey(region.getRegionInfo().getStartKey()).
+ setEndKey(Bytes.toBytes("002")).setSplit(false).
+ setRegionId(region.getRegionInfo().getRegionId() +
+ EnvironmentEdgeManager.currentTime()).
+ build();
+ RegionInfo daughterB = RegionInfoBuilder.newBuilder(table).setStartKey(Bytes.toBytes("002"))
+ .setEndKey(region.getRegionInfo().getEndKey()).setSplit(false)
+ .setRegionId(region.getRegionInfo().getRegionId()).build();
+ HStoreFile file = (HStoreFile) region.getStore(FAMILY_NAME).getStorefiles().toArray()[0];
+ List<Path> splitFilesA = new ArrayList<>();
+ splitFilesA.add(regionFS
+ .splitStoreFile(daughterA, Bytes.toString(FAMILY_NAME), file,
+ Bytes.toBytes("002"), false, region.getSplitPolicy()));
+ List<Path> splitFilesB = new ArrayList<>();
+ splitFilesB.add(regionFS
+ .splitStoreFile(daughterB, Bytes.toString(FAMILY_NAME), file,
+ Bytes.toBytes("002"), true, region.getSplitPolicy()));
+ MasterProcedureEnv env = TEST_UTIL.getMiniHBaseCluster().getMaster().
+ getMasterProcedureExecutor().getEnvironment();
+ Path resultA = regionFS.commitDaughterRegion(daughterA, splitFilesA, env);
+ Path resultB = regionFS.commitDaughterRegion(daughterB, splitFilesB, env);
+ FileSystem fs = regionFS.getFileSystem();
+ verifyFilesAreTracked(resultA, fs);
+ verifyFilesAreTracked(resultB, fs);
+ }
+
+ @Test
+ public void testCommitMergedRegion() throws Exception {
+ TableName table = TableName.valueOf(name.getMethodName());
+ TEST_UTIL.createTable(table, FAMILY_NAME);
+ //splitting the table first
+ TEST_UTIL.getAdmin().split(table, Bytes.toBytes("002"));
+ //Add data and flush to create files in the two different regions
+ putThreeRowsAndFlush(table);
+ List<HRegion> regions = TEST_UTIL.getHBaseCluster().getRegions(table);
+ HRegion first = regions.get(0);
+ HRegion second = regions.get(1);
+ HRegionFileSystem regionFS = first.getRegionFileSystem();
+
+ RegionInfo mergeResult =
+ RegionInfoBuilder.newBuilder(table).setStartKey(first.getRegionInfo().getStartKey())
+ .setEndKey(second.getRegionInfo().getEndKey()).setSplit(false)
+ .setRegionId(first.getRegionInfo().getRegionId() +
+ EnvironmentEdgeManager.currentTime()).build();
+
+ HRegionFileSystem mergeFS = HRegionFileSystem.createRegionOnFileSystem(
+ TEST_UTIL.getHBaseCluster().getMaster().getConfiguration(),
+ regionFS.getFileSystem(), regionFS.getTableDir(), mergeResult);
+
+ List<Path> mergedFiles = new ArrayList<>();
+ //merge file from first region
+ mergedFiles.add(mergeFileFromRegion(first, mergeFS));
+ //merge file from second region
+ mergedFiles.add(mergeFileFromRegion(second, mergeFS));
+ MasterProcedureEnv env = TEST_UTIL.getMiniHBaseCluster().getMaster().
+ getMasterProcedureExecutor().getEnvironment();
+ mergeFS.commitMergedRegion(mergedFiles, env);
+ //validate
+ FileSystem fs = first.getRegionFileSystem().getFileSystem();
+ Path finalMergeDir = new Path(first.getRegionFileSystem().getTableDir(),
+ mergeResult.getEncodedName());
+ verifyFilesAreTracked(finalMergeDir, fs);
+ }
+
+ @Test
+ public void testSplitLoadsFromTracker() throws Exception {
+ TableName table = TableName.valueOf(name.getMethodName());
+ TEST_UTIL.createTable(table, FAMILY_NAME);
+ //Add data and flush to create files in the two different regions
+ putThreeRowsAndFlush(table);
+ HRegion region = TEST_UTIL.getHBaseCluster().getRegions(table).get(0);
+ Pair<StoreFileInfo, String> copyResult = copyFileInTheStoreDir(region);
+ StoreFileInfo fileInfo = copyResult.getFirst();
+ String copyName = copyResult.getSecond();
+ //Now splits the region
+ TEST_UTIL.getAdmin().split(table, Bytes.toBytes("002"));
+ List<HRegion> regions = TEST_UTIL.getHBaseCluster().getRegions(table);
+ HRegion first = regions.get(0);
+ validateDaughterRegionsFiles(first, fileInfo.getActiveFileName(), copyName);
+ HRegion second = regions.get(1);
+ validateDaughterRegionsFiles(second, fileInfo.getActiveFileName(), copyName);
+ }
+
+ @Test
+ public void testMergeLoadsFromTracker() throws Exception {
+ TableName table = TableName.valueOf(name.getMethodName());
+ TEST_UTIL.createTable(table, new byte[][]{FAMILY_NAME},
+ new byte[][]{Bytes.toBytes("002")});
+ //Add data and flush to create files in the two different regions
+ putThreeRowsAndFlush(table);
+ List<HRegion> regions = TEST_UTIL.getHBaseCluster().getRegions(table);
+ HRegion first = regions.get(0);
+ Pair<StoreFileInfo, String> copyResult = copyFileInTheStoreDir(first);
+ StoreFileInfo fileInfo = copyResult.getFirst();
+ String copyName = copyResult.getSecond();
+ //Now merges the first two regions
+ TEST_UTIL.getAdmin().mergeRegionsAsync(new byte[][]{
+ first.getRegionInfo().getEncodedNameAsBytes(),
+ regions.get(1).getRegionInfo().getEncodedNameAsBytes()
+ }, true).get(10, TimeUnit.SECONDS);
+ regions = TEST_UTIL.getHBaseCluster().getRegions(table);
+ HRegion merged = regions.get(0);
+ validateDaughterRegionsFiles(merged, fileInfo.getActiveFileName(), copyName);
+ }
+
+ private Pair<StoreFileInfo,String> copyFileInTheStoreDir(HRegion region) throws IOException {
+ Path storeDir = region.getRegionFileSystem().getStoreDir("info");
+ //gets the single file
+ StoreFileInfo fileInfo = region.getRegionFileSystem().getStoreFiles("info").get(0);
+ //make a copy of the valid file staight into the store dir, so that it's not tracked.
+ String copyName = UUID.randomUUID().toString().replaceAll("-", "");
+ Path copy = new Path(storeDir, copyName);
+ FileUtil.copy(region.getFilesystem(), fileInfo.getFileStatus(), region.getFilesystem(),
+ copy , false, false, TEST_UTIL.getConfiguration());
+ return new Pair<>(fileInfo, copyName);
+ }
+
+ private void validateDaughterRegionsFiles(HRegion region, String orignalFileName,
+ String untrackedFile) throws IOException {
+ //verify there's no link for the untracked, copied file in first region
+ List<StoreFileInfo> infos = region.getRegionFileSystem().getStoreFiles("info");
+ final MutableBoolean foundLink = new MutableBoolean(false);
+ infos.stream().forEach(i -> {
+ i.getActiveFileName().contains(orignalFileName);
+ if(i.getActiveFileName().contains(untrackedFile)){
+ fail();
+ }
+ if(i.getActiveFileName().contains(orignalFileName)){
+ foundLink.setTrue();
+ }
+ });
+ assertTrue(foundLink.booleanValue());
+ }
+
+ private void verifyFilesAreTracked(Path regionDir, FileSystem fs) throws Exception {
+ String storeId = regionDir.getName() + "-info";
+ for(FileStatus f : fs.listStatus(new Path(regionDir, Bytes.toString(FAMILY_NAME)))){
+ assertTrue(TestStoreFileTracker.trackedFiles.get(storeId).stream().filter(s ->
+ s.getPath().equals(f.getPath())).findFirst().isPresent());
+ }
+ }
+
+ private Path mergeFileFromRegion(HRegion regionToMerge, HRegionFileSystem mergeFS)
+ throws IOException {
+ HStoreFile file = (HStoreFile) regionToMerge.getStore(FAMILY_NAME).getStorefiles().toArray()[0];
+ return mergeFS.mergeStoreFile(regionToMerge.getRegionInfo(), Bytes.toString(FAMILY_NAME), file);
+ }
+
+ private void putThreeRowsAndFlush(TableName table) throws IOException {
+ Table tbl = TEST_UTIL.getConnection().getTable(table);
+ Put put = new Put(Bytes.toBytes("001"));
+ byte[] qualifier = Bytes.toBytes("1");
+ put.addColumn(FAMILY_NAME, qualifier, Bytes.toBytes(1));
+ tbl.put(put);
+ put = new Put(Bytes.toBytes("002"));
+ put.addColumn(FAMILY_NAME, qualifier, Bytes.toBytes(2));
+ tbl.put(put);
+ put = new Put(Bytes.toBytes("003"));
+ put.addColumn(FAMILY_NAME, qualifier, Bytes.toBytes(2));
+ tbl.put(put);
+ TEST_UTIL.flush(table);
+ }
+}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStripeStoreEngine.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStripeStoreEngine.java
index eb0b1c1..80012df 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStripeStoreEngine.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStripeStoreEngine.java
@@ -120,6 +120,7 @@ public class TestStripeStoreEngine {
private static TestStoreEngine createEngine(Configuration conf) throws Exception {
HRegion region = mock(HRegion.class);
HStore store = mock(HStore.class);
+ store.conf = conf;
when(store.getRegionInfo()).thenReturn(RegionInfoBuilder.FIRST_META_REGIONINFO);
when(store.getHRegion()).thenReturn(region);
CellComparatorImpl kvComparator = mock(CellComparatorImpl.class);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestStoreFileTracker.java
similarity index 61%
copy from hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
copy to hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestStoreFileTracker.java
index fa04481..05ca1fc 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestStoreFileTracker.java
@@ -18,43 +18,39 @@
package org.apache.hadoop.hbase.regionserver.storefiletracker;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.regionserver.StoreContext;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
-import org.apache.yetus.audience.InterfaceAudience;
-
-/**
- * The default implementation for store file tracker, where we do not persist the store file list,
- * and use listing when loading store files.
- */
-@InterfaceAudience.Private
-class DefaultStoreFileTracker extends StoreFileTrackerBase {
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
- public DefaultStoreFileTracker(Configuration conf, boolean isPrimaryReplica,
- StoreContext ctx) {
- super(conf, isPrimaryReplica, ctx);
- }
+public class TestStoreFileTracker extends DefaultStoreFileTracker {
- @Override
- public List<StoreFileInfo> load() throws IOException {
- return ctx.getRegionFileSystem().getStoreFiles(ctx.getFamily().getNameAsString());
- }
+ private static final Logger LOG = LoggerFactory.getLogger(TestStoreFileTracker.class);
+ public static Map<String, List<StoreFileInfo>> trackedFiles = new HashMap<>();
+ private String storeId;
- @Override
- public boolean requireWritingToTmpDirFirst() {
- return true;
+ public TestStoreFileTracker(Configuration conf, boolean isPrimaryReplica, StoreContext ctx) {
+ super(conf, isPrimaryReplica, ctx);
+ this.storeId = ctx.getRegionInfo().getEncodedName() + "-" + ctx.getFamily().getNameAsString();
+ LOG.info("created storeId: {}", storeId);
+ trackedFiles.computeIfAbsent(storeId, v -> new ArrayList<>());
}
@Override
protected void doAddNewStoreFiles(Collection<StoreFileInfo> newFiles) throws IOException {
- // NOOP
+ LOG.info("adding to storeId: {}", storeId);
+ trackedFiles.get(storeId).addAll(newFiles);
}
@Override
- protected void doAddCompactionResults(Collection<StoreFileInfo> compactedFiles,
- Collection<StoreFileInfo> newFiles) throws IOException {
- // NOOP
+ public List<StoreFileInfo> load() throws IOException {
+ return trackedFiles.get(storeId);
}
}
[hbase] 02/05: HBASE-25988 Store the store file list by a file
(#3578)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-26067
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 488c4cf7dc904f944a1cc3321278ae83cb542c9a
Author: Duo Zhang <zh...@apache.org>
AuthorDate: Thu Aug 26 18:51:12 2021 +0800
HBASE-25988 Store the store file list by a file (#3578)
Signed-off-by: Wellington Chevreuil <wc...@apache.org>
---
.../protobuf/server/region/StoreFileTracker.proto | 29 +++--
.../hadoop/hbase/regionserver/StoreContext.java | 5 +
.../hadoop/hbase/regionserver/StoreEngine.java | 8 +-
.../storefiletracker/DefaultStoreFileTracker.java | 5 +-
.../FileBasedStoreFileTracker.java | 142 +++++++++++++++++++++
.../storefiletracker/StoreFileListFile.java | 142 +++++++++++++++++++++
.../storefiletracker/StoreFileTrackerBase.java | 11 +-
.../storefiletracker/StoreFileTrackerFactory.java | 12 +-
.../TestRegionWithFileBasedStoreFileTracker.java | 109 ++++++++++++++++
9 files changed, 430 insertions(+), 33 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java b/hbase-protocol-shaded/src/main/protobuf/server/region/StoreFileTracker.proto
similarity index 57%
copy from hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
copy to hbase-protocol-shaded/src/main/protobuf/server/region/StoreFileTracker.proto
index 4f7231b..2a269ea 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
+++ b/hbase-protocol-shaded/src/main/protobuf/server/region/StoreFileTracker.proto
@@ -15,21 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.hadoop.hbase.regionserver.storefiletracker;
+syntax = "proto2";
+// This file contains protocol buffers that are used for store file tracker.
+package hbase.pb;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.regionserver.StoreContext;
-import org.apache.yetus.audience.InterfaceAudience;
+option java_package = "org.apache.hadoop.hbase.shaded.protobuf.generated";
+option java_outer_classname = "StoreFileTrackerProtos";
+option java_generic_services = true;
+option java_generate_equals_and_hash = true;
+option optimize_for = SPEED;
-/**
- * Factory method for creating store file tracker.
- */
-@InterfaceAudience.Private
-public final class StoreFileTrackerFactory {
+message StoreFileEntry {
+ required string name = 1;
+ required uint64 size = 2;
+}
- public static StoreFileTracker create(Configuration conf, TableName tableName,
- boolean isPrimaryReplica, StoreContext ctx) {
- return new DefaultStoreFileTracker(conf, tableName, isPrimaryReplica, ctx);
- }
+message StoreFileList {
+ required uint64 timestamp = 1;
+ repeated StoreFileEntry store_file = 2;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreContext.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreContext.java
index 2a9f968..588f8f4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreContext.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreContext.java
@@ -22,6 +22,7 @@ import java.util.Collection;
import java.util.function.Supplier;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CellComparator;
+import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.io.HeapSize;
@@ -109,6 +110,10 @@ public final class StoreContext implements HeapSize {
return coprocessorHost;
}
+ public TableName getTableName() {
+ return getRegionInfo().getTable();
+ }
+
public RegionInfo getRegionInfo() {
return regionFileSystem.getRegionInfo();
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java
index f9d6c29..61b7dfa 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java
@@ -172,9 +172,9 @@ public abstract class StoreEngine<SF extends StoreFlusher, CP extends Compaction
return this.storeFlusher;
}
- private StoreFileTracker createStoreFileTracker(HStore store) {
- return StoreFileTrackerFactory.create(store.conf, store.getRegionInfo().getTable(),
- store.isPrimaryReplicaStore(), store.getStoreContext());
+ private StoreFileTracker createStoreFileTracker(Configuration conf, HStore store) {
+ return StoreFileTrackerFactory.create(conf, store.isPrimaryReplicaStore(),
+ store.getStoreContext());
}
/**
@@ -205,7 +205,7 @@ public abstract class StoreEngine<SF extends StoreFlusher, CP extends Compaction
this.ctx = store.getStoreContext();
this.coprocessorHost = store.getHRegion().getCoprocessorHost();
this.openStoreFileThreadPoolCreator = store.getHRegion()::getStoreFileOpenAndCloseThreadPool;
- this.storeFileTracker = createStoreFileTracker(store);
+ this.storeFileTracker = createStoreFileTracker(conf, store);
assert compactor != null && compactionPolicy != null && storeFileManager != null &&
storeFlusher != null && storeFileTracker != null;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
index d4c9a86..fa04481 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
@@ -21,7 +21,6 @@ import java.io.IOException;
import java.util.Collection;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.regionserver.StoreContext;
import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
import org.apache.yetus.audience.InterfaceAudience;
@@ -33,9 +32,9 @@ import org.apache.yetus.audience.InterfaceAudience;
@InterfaceAudience.Private
class DefaultStoreFileTracker extends StoreFileTrackerBase {
- public DefaultStoreFileTracker(Configuration conf, TableName tableName, boolean isPrimaryReplica,
+ public DefaultStoreFileTracker(Configuration conf, boolean isPrimaryReplica,
StoreContext ctx) {
- super(conf, tableName, isPrimaryReplica, ctx);
+ super(conf, isPrimaryReplica, ctx);
}
@Override
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FileBasedStoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FileBasedStoreFileTracker.java
new file mode 100644
index 0000000..de28b0e
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FileBasedStoreFileTracker.java
@@ -0,0 +1,142 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.regionserver.StoreContext;
+import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.apache.hadoop.hbase.shaded.protobuf.generated.StoreFileTrackerProtos.StoreFileEntry;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.StoreFileTrackerProtos.StoreFileList;
+
+/**
+ * A file based store file tracker.
+ * <p/>
+ * For this tracking way, the store file list will be persistent into a file, so we can write the
+ * new store files directly to the final data directory, as we will not load the broken files. This
+ * will greatly reduce the time for flush and compaction on some object storages as a rename is
+ * actual a copy on them. And it also avoid listing when loading store file list, which could also
+ * speed up the loading of store files as listing is also not a fast operation on most object
+ * storages.
+ */
+@InterfaceAudience.Private
+public class FileBasedStoreFileTracker extends StoreFileTrackerBase {
+
+ private final StoreFileListFile backedFile;
+
+ private final Map<String, StoreFileInfo> storefiles = new HashMap<>();
+
+ public FileBasedStoreFileTracker(Configuration conf, boolean isPrimaryReplica, StoreContext ctx) {
+ super(conf, isPrimaryReplica, ctx);
+ backedFile = new StoreFileListFile(ctx);
+ }
+
+ @Override
+ public List<StoreFileInfo> load() throws IOException {
+ StoreFileList list = backedFile.load();
+ if (list == null) {
+ return Collections.emptyList();
+ }
+ FileSystem fs = ctx.getRegionFileSystem().getFileSystem();
+ List<StoreFileInfo> infos = new ArrayList<>();
+ for (StoreFileEntry entry : list.getStoreFileList()) {
+ infos.add(ServerRegionReplicaUtil.getStoreFileInfo(conf, fs, ctx.getRegionInfo(),
+ ctx.getRegionFileSystem().getRegionInfoForFS(), ctx.getFamily().getNameAsString(),
+ new Path(ctx.getFamilyStoreDirectoryPath(), entry.getName())));
+ }
+ // In general, for primary replica, the load method should only be called once when
+ // initialization, so we do not need synchronized here. And for secondary replicas, though the
+ // load method could be called multiple times, we will never call other methods so no
+ // synchronized is also fine.
+ // But we have a refreshStoreFiles method in the Region interface, which can be called by CPs,
+ // and we have a RefreshHFilesEndpoint example to expose the refreshStoreFiles method as RPC, so
+ // for safety, let's still keep the synchronized here.
+ synchronized (storefiles) {
+ for (StoreFileInfo info : infos) {
+ storefiles.put(info.getPath().getName(), info);
+ }
+ }
+ return infos;
+ }
+
+ @Override
+ protected boolean requireWritingToTmpDirFirst() {
+ return false;
+ }
+
+ private StoreFileEntry toStoreFileEntry(StoreFileInfo info) {
+ return StoreFileEntry.newBuilder().setName(info.getPath().getName()).setSize(info.getSize())
+ .build();
+ }
+
+ @Override
+ protected void doAddNewStoreFiles(Collection<StoreFileInfo> newFiles) throws IOException {
+ synchronized (storefiles) {
+ StoreFileList.Builder builder = StoreFileList.newBuilder();
+ for (StoreFileInfo info : storefiles.values()) {
+ builder.addStoreFile(toStoreFileEntry(info));
+ }
+ for (StoreFileInfo info : newFiles) {
+ builder.addStoreFile(toStoreFileEntry(info));
+ }
+ backedFile.update(builder);
+ for (StoreFileInfo info : newFiles) {
+ storefiles.put(info.getPath().getName(), info);
+ }
+ }
+ }
+
+ @Override
+ protected void doAddCompactionResults(Collection<StoreFileInfo> compactedFiles,
+ Collection<StoreFileInfo> newFiles) throws IOException {
+ Set<String> compactedFileNames =
+ compactedFiles.stream().map(info -> info.getPath().getName()).collect(Collectors.toSet());
+ synchronized (storefiles) {
+ StoreFileList.Builder builder = StoreFileList.newBuilder();
+ storefiles.forEach((name, info) -> {
+ if (compactedFileNames.contains(name)) {
+ return;
+ }
+ builder.addStoreFile(toStoreFileEntry(info));
+ });
+ for (StoreFileInfo info : newFiles) {
+ builder.addStoreFile(toStoreFileEntry(info));
+ }
+ backedFile.update(builder);
+ for (String name : compactedFileNames) {
+ storefiles.remove(name);
+ }
+ for (StoreFileInfo info : newFiles) {
+ storefiles.put(info.getPath().getName(), info);
+ }
+ }
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java
new file mode 100644
index 0000000..c778bfc
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java
@@ -0,0 +1,142 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.regionserver.StoreContext;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
+import org.apache.hbase.thirdparty.com.google.common.io.ByteStreams;
+import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
+
+import org.apache.hadoop.hbase.shaded.protobuf.generated.StoreFileTrackerProtos.StoreFileList;
+
+/**
+ * To fully avoid listing, here we use two files for tracking. When loading, we will try to read
+ * both the two files, if only one exists, we will trust this one, if both exist, we will compare
+ * the timestamp to see which one is newer and trust that one. And we will record in memory that
+ * which one is trusted by us, and when we need to update the store file list, we will write to the
+ * other file.
+ * <p/>
+ * So in this way, we could avoid listing when we want to load the store file list file.
+ */
+@InterfaceAudience.Private
+class StoreFileListFile {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StoreFileListFile.class);
+
+ private static final String TRACK_FILE_DIR = ".filelist";
+
+ private static final String TRACK_FILE = "f1";
+
+ private static final String TRACK_FILE_ROTATE = "f2";
+
+ private final StoreContext ctx;
+
+ private final Path trackFileDir;
+
+ private final Path[] trackFiles = new Path[2];
+
+ // this is used to make sure that we do not go backwards
+ private long prevTimestamp = -1;
+
+ private int nextTrackFile = -1;
+
+ StoreFileListFile(StoreContext ctx) {
+ this.ctx = ctx;
+ trackFileDir = new Path(ctx.getFamilyStoreDirectoryPath(), TRACK_FILE_DIR);
+ trackFiles[0] = new Path(trackFileDir, TRACK_FILE);
+ trackFiles[1] = new Path(trackFileDir, TRACK_FILE_ROTATE);
+ }
+
+ private StoreFileList load(Path path) throws IOException {
+ FileSystem fs = ctx.getRegionFileSystem().getFileSystem();
+ byte[] bytes;
+ try (FSDataInputStream in = fs.open(path)) {
+ bytes = ByteStreams.toByteArray(in);
+ }
+ // Read all the bytes and then parse it, so we will only throw InvalidProtocolBufferException
+ // here. This is very important for upper layer to determine whether this is the normal case,
+ // where the file does not exist or is incomplete. If there is another type of exception, the
+ // upper layer should throw it out instead of just ignoring it, otherwise it will lead to data
+ // loss.
+ return StoreFileList.parseFrom(bytes);
+ }
+
+ private int select(StoreFileList[] lists) {
+ if (lists[0] == null) {
+ return 1;
+ }
+ if (lists[1] == null) {
+ return 0;
+ }
+ return lists[0].getTimestamp() >= lists[1].getTimestamp() ? 0 : 1;
+ }
+
+ StoreFileList load() throws IOException {
+ StoreFileList[] lists = new StoreFileList[2];
+ for (int i = 0; i < 2; i++) {
+ try {
+ lists[i] = load(trackFiles[i]);
+ } catch (FileNotFoundException | InvalidProtocolBufferException e) {
+ // this is normal case, so use info and do not log stacktrace
+ LOG.info("Failed to load track file {}: {}", trackFiles[i], e);
+ }
+ }
+ int winnerIndex = select(lists);
+ if (lists[winnerIndex] != null) {
+ nextTrackFile = 1 - winnerIndex;
+ prevTimestamp = lists[winnerIndex].getTimestamp();
+ } else {
+ nextTrackFile = 0;
+ }
+ return lists[winnerIndex];
+ }
+
+ /**
+ * We will set the timestamp in this method so just pass the builder in
+ */
+ void update(StoreFileList.Builder builder) throws IOException {
+ Preconditions.checkState(nextTrackFile >= 0, "should call load first before calling update");
+ FileSystem fs = ctx.getRegionFileSystem().getFileSystem();
+ long timestamp = Math.max(prevTimestamp + 1, EnvironmentEdgeManager.currentTime());
+ try (FSDataOutputStream out = fs.create(trackFiles[nextTrackFile], true)) {
+ builder.setTimestamp(timestamp).build().writeTo(out);
+ }
+ // record timestamp
+ prevTimestamp = timestamp;
+ // rotate the file
+ nextTrackFile = 1 - nextTrackFile;
+ try {
+ fs.delete(trackFiles[nextTrackFile], false);
+ } catch (IOException e) {
+ // we will create new file with overwrite = true, so not a big deal here, only for speed up
+ // loading as we do not need to read this file when loading(we will hit FileNotFoundException)
+ LOG.debug("failed to delete old track file {}, not a big deal, just ignore", e);
+ }
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
index 2451f45..92c6992 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
@@ -21,7 +21,6 @@ import java.io.IOException;
import java.util.Collection;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.crypto.Encryption;
@@ -51,18 +50,14 @@ abstract class StoreFileTrackerBase implements StoreFileTracker {
protected final Configuration conf;
- protected final TableName tableName;
-
protected final boolean isPrimaryReplica;
protected final StoreContext ctx;
private volatile boolean cacheOnWriteLogged;
- protected StoreFileTrackerBase(Configuration conf, TableName tableName, boolean isPrimaryReplica,
- StoreContext ctx) {
+ protected StoreFileTrackerBase(Configuration conf, boolean isPrimaryReplica, StoreContext ctx) {
this.conf = conf;
- this.tableName = tableName;
this.isPrimaryReplica = isPrimaryReplica;
this.ctx = ctx;
}
@@ -95,7 +90,7 @@ abstract class StoreFileTrackerBase implements StoreFileTracker {
.withBlockSize(family.getBlocksize()).withHBaseCheckSum(true)
.withDataBlockEncoding(family.getDataBlockEncoding()).withEncryptionContext(encryptionContext)
.withCreateTime(EnvironmentEdgeManager.currentTime()).withColumnFamily(family.getName())
- .withTableName(tableName.getName()).withCellComparator(ctx.getComparator()).build();
+ .withTableName(ctx.getTableName().getName()).withCellComparator(ctx.getComparator()).build();
return hFileContext;
}
@@ -153,7 +148,7 @@ abstract class StoreFileTrackerBase implements StoreFileTracker {
outputDir =
new Path(ctx.getRegionFileSystem().getTempDir(), ctx.getFamily().getNameAsString());
} else {
- throw new UnsupportedOperationException("not supported yet");
+ outputDir = ctx.getFamilyStoreDirectoryPath();
}
StoreFileWriter.Builder builder =
new StoreFileWriter.Builder(conf, writerCacheConf, ctx.getRegionFileSystem().getFileSystem())
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
index 4f7231b..6cdfaf4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
@@ -18,8 +18,8 @@
package org.apache.hadoop.hbase.regionserver.storefiletracker;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.regionserver.StoreContext;
+import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.yetus.audience.InterfaceAudience;
/**
@@ -28,8 +28,12 @@ import org.apache.yetus.audience.InterfaceAudience;
@InterfaceAudience.Private
public final class StoreFileTrackerFactory {
- public static StoreFileTracker create(Configuration conf, TableName tableName,
- boolean isPrimaryReplica, StoreContext ctx) {
- return new DefaultStoreFileTracker(conf, tableName, isPrimaryReplica, ctx);
+ public static final String TRACK_IMPL = "hbase.store.file-tracker.impl";
+
+ public static StoreFileTracker create(Configuration conf, boolean isPrimaryReplica,
+ StoreContext ctx) {
+ Class<? extends StoreFileTracker> tracker =
+ conf.getClass(TRACK_IMPL, DefaultStoreFileTracker.class, StoreFileTracker.class);
+ return ReflectionUtils.newInstance(tracker, conf, isPrimaryReplica, ctx);
}
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestRegionWithFileBasedStoreFileTracker.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestRegionWithFileBasedStoreFileTracker.java
new file mode 100644
index 0000000..3bc60d1
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestRegionWithFileBasedStoreFileTracker.java
@@ -0,0 +1,109 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.RegionInfoBuilder;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+
+@Category({ RegionServerTests.class, MediumTests.class })
+public class TestRegionWithFileBasedStoreFileTracker {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestRegionWithFileBasedStoreFileTracker.class);
+
+ private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
+
+ private static final byte[] CF = Bytes.toBytes("cf");
+
+ private static final byte[] CQ = Bytes.toBytes("cq");
+
+ private static final TableDescriptor TD =
+ TableDescriptorBuilder.newBuilder(TableName.valueOf("file_based_tracker"))
+ .setColumnFamily(ColumnFamilyDescriptorBuilder.of(CF)).build();
+
+ private static final RegionInfo RI = RegionInfoBuilder.newBuilder(TD.getTableName()).build();
+
+ @Rule
+ public TestName name = new TestName();
+
+ private HRegion region;
+
+ @Before
+ public void setUp() throws IOException {
+ Configuration conf = new Configuration(UTIL.getConfiguration());
+ conf.setClass(StoreFileTrackerFactory.TRACK_IMPL, FileBasedStoreFileTracker.class,
+ StoreFileTracker.class);
+ region =
+ HBaseTestingUtil.createRegionAndWAL(RI, UTIL.getDataTestDir(name.getMethodName()), conf, TD);
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ if (region != null) {
+ HBaseTestingUtil.closeRegionAndWAL(region);
+ }
+ UTIL.cleanupTestDir();
+ }
+
+ @Test
+ public void testFlushAndCompaction() throws IOException {
+ for (int i = 0; i < 10; i++) {
+ for (int j = 0; j < 10; j++) {
+ int v = i * 10 + j;
+ region.put(new Put(Bytes.toBytes(v)).addColumn(CF, CQ, Bytes.toBytes(v)));
+ }
+ region.flush(true);
+ if (i % 3 == 2) {
+ region.compact(true);
+ }
+ }
+ // reopen the region, make sure the store file tracker works, i.e, we can get all the records
+ // back
+ region.close();
+ region = HRegion.openHRegion(region, null);
+ for (int i = 0; i < 100; i++) {
+ Result result = region.get(new Get(Bytes.toBytes(i)));
+ assertEquals(i, Bytes.toInt(result.getValue(CF, CQ)));
+ }
+ }
+}
[hbase] 04/05: HBASE-26224 Introduce a MigrationStoreFileTracker to
support migrating from different store file tracker implementations (#3656)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-26067
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 24f48b06952c22a0c69bb39694883d2ce00a74a7
Author: Duo Zhang <zh...@apache.org>
AuthorDate: Thu Sep 9 12:11:25 2021 +0800
HBASE-26224 Introduce a MigrationStoreFileTracker to support migrating from different store file tracker implementations (#3656)
Signed-off-by: Wellington Chevreuil <wc...@apache.org>
---
.../storefiletracker/DefaultStoreFileTracker.java | 10 +-
.../FileBasedStoreFileTracker.java | 15 +-
.../MigrationStoreFileTracker.java | 88 ++++++++++
.../storefiletracker/StoreFileListFile.java | 6 +-
.../storefiletracker/StoreFileTrackerBase.java | 12 +-
.../storefiletracker/StoreFileTrackerFactory.java | 39 +++--
.../TestMigrationStoreFileTracker.java | 193 +++++++++++++++++++++
7 files changed, 343 insertions(+), 20 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
index 22e0513..a13b75b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/DefaultStoreFileTracker.java
@@ -19,6 +19,7 @@ package org.apache.hadoop.hbase.regionserver.storefiletracker;
import java.io.IOException;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
@@ -39,7 +40,9 @@ class DefaultStoreFileTracker extends StoreFileTrackerBase {
@Override
public List<StoreFileInfo> load() throws IOException {
- return ctx.getRegionFileSystem().getStoreFiles(ctx.getFamily().getNameAsString());
+ List<StoreFileInfo> files =
+ ctx.getRegionFileSystem().getStoreFiles(ctx.getFamily().getNameAsString());
+ return files != null ? files : Collections.emptyList();
}
@Override
@@ -57,4 +60,9 @@ class DefaultStoreFileTracker extends StoreFileTrackerBase {
Collection<StoreFileInfo> newFiles) throws IOException {
// NOOP
}
+
+ @Override
+ void set(List<StoreFileInfo> files) {
+ // NOOP
+ }
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FileBasedStoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FileBasedStoreFileTracker.java
index de28b0e..c370b87 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FileBasedStoreFileTracker.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/FileBasedStoreFileTracker.java
@@ -48,7 +48,7 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.StoreFileTrackerProtos.
* storages.
*/
@InterfaceAudience.Private
-public class FileBasedStoreFileTracker extends StoreFileTrackerBase {
+class FileBasedStoreFileTracker extends StoreFileTrackerBase {
private final StoreFileListFile backedFile;
@@ -139,4 +139,17 @@ public class FileBasedStoreFileTracker extends StoreFileTrackerBase {
}
}
}
+
+ @Override
+ void set(List<StoreFileInfo> files) throws IOException {
+ synchronized (storefiles) {
+ storefiles.clear();
+ StoreFileList.Builder builder = StoreFileList.newBuilder();
+ for (StoreFileInfo info : files) {
+ storefiles.put(info.getPath().getName(), info);
+ builder.addStoreFile(toStoreFileEntry(info));
+ }
+ backedFile.update(builder);
+ }
+ }
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/MigrationStoreFileTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/MigrationStoreFileTracker.java
new file mode 100644
index 0000000..e486e6d
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/MigrationStoreFileTracker.java
@@ -0,0 +1,88 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.regionserver.StoreContext;
+import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
+
+/**
+ * A store file tracker used for migrating between store file tracker implementations.
+ */
+@InterfaceAudience.Private
+class MigrationStoreFileTracker extends StoreFileTrackerBase {
+
+ public static final String SRC_IMPL = "hbase.store.file-tracker.migration.src.impl";
+
+ public static final String DST_IMPL = "hbase.store.file-tracker.migration.dst.impl";
+
+ private final StoreFileTrackerBase src;
+
+ private final StoreFileTrackerBase dst;
+
+ public MigrationStoreFileTracker(Configuration conf, boolean isPrimaryReplica, StoreContext ctx) {
+ super(conf, isPrimaryReplica, ctx);
+ this.src = StoreFileTrackerFactory.create(conf, SRC_IMPL, isPrimaryReplica, ctx);
+ this.dst = StoreFileTrackerFactory.create(conf, DST_IMPL, isPrimaryReplica, ctx);
+ Preconditions.checkArgument(!src.getClass().equals(dst.getClass()),
+ "src and dst is the same: %s", src.getClass());
+ }
+
+ @Override
+ public List<StoreFileInfo> load() throws IOException {
+ List<StoreFileInfo> files = src.load();
+ dst.set(files);
+ return files;
+ }
+
+ @Override
+ protected boolean requireWritingToTmpDirFirst() {
+ // Returns true if either of the two StoreFileTracker returns true.
+ // For example, if we want to migrate from a tracker implementation which can ignore the broken
+ // files under data directory to a tracker implementation which can not, if we still allow
+ // writing in tmp directory directly, we may have some broken files under the data directory and
+ // then after we finally change the implementation which can not ignore the broken files, we
+ // will be in trouble.
+ return src.requireWritingToTmpDirFirst() || dst.requireWritingToTmpDirFirst();
+ }
+
+ @Override
+ protected void doAddNewStoreFiles(Collection<StoreFileInfo> newFiles) throws IOException {
+ src.doAddNewStoreFiles(newFiles);
+ dst.doAddNewStoreFiles(newFiles);
+ }
+
+ @Override
+ protected void doAddCompactionResults(Collection<StoreFileInfo> compactedFiles,
+ Collection<StoreFileInfo> newFiles) throws IOException {
+ src.doAddCompactionResults(compactedFiles, newFiles);
+ dst.doAddCompactionResults(compactedFiles, newFiles);
+ }
+
+ @Override
+ void set(List<StoreFileInfo> files) {
+ throw new UnsupportedOperationException(
+ "Should not call this method on " + getClass().getSimpleName());
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java
index c778bfc..ffb3647 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileListFile.java
@@ -29,7 +29,6 @@ import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
import org.apache.hbase.thirdparty.com.google.common.io.ByteStreams;
import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
@@ -121,7 +120,10 @@ class StoreFileListFile {
* We will set the timestamp in this method so just pass the builder in
*/
void update(StoreFileList.Builder builder) throws IOException {
- Preconditions.checkState(nextTrackFile >= 0, "should call load first before calling update");
+ if (nextTrackFile < 0) {
+ // we need to call load first to load the prevTimestamp and also the next file
+ load();
+ }
FileSystem fs = ctx.getRegionFileSystem().getFileSystem();
long timestamp = Math.max(prevTimestamp + 1, EnvironmentEdgeManager.currentTime());
try (FSDataOutputStream out = fs.create(trackFiles[nextTrackFile], true)) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
index 92c6992..d860f8e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerBase.java
@@ -19,6 +19,7 @@ package org.apache.hadoop.hbase.regionserver.storefiletracker;
import java.io.IOException;
import java.util.Collection;
+import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
@@ -95,8 +96,7 @@ abstract class StoreFileTrackerBase implements StoreFileTracker {
}
@Override
- public final StoreFileWriter createWriter(CreateStoreFileWriterParams params)
- throws IOException {
+ public final StoreFileWriter createWriter(CreateStoreFileWriterParams params) throws IOException {
if (!isPrimaryReplica) {
throw new IllegalStateException("Should not call create writer on secondary replicas");
}
@@ -170,4 +170,12 @@ abstract class StoreFileTrackerBase implements StoreFileTracker {
protected abstract void doAddCompactionResults(Collection<StoreFileInfo> compactedFiles,
Collection<StoreFileInfo> newFiles) throws IOException;
+
+ /**
+ * used to mirror the store file list after loading when migration.
+ * <p/>
+ * Do not add this method to the {@link StoreFileTracker} interface since we do not need this
+ * method in upper layer.
+ */
+ abstract void set(List<StoreFileInfo> files) throws IOException;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
index c446d5a..2c2b71d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/storefiletracker/StoreFileTrackerFactory.java
@@ -30,6 +30,8 @@ import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
+
/**
* Factory method for creating store file tracker.
*/
@@ -39,7 +41,7 @@ public final class StoreFileTrackerFactory {
private static final Logger LOG = LoggerFactory.getLogger(StoreFileTrackerFactory.class);
public static StoreFileTracker create(Configuration conf, boolean isPrimaryReplica,
- StoreContext ctx) {
+ StoreContext ctx) {
Class<? extends StoreFileTracker> tracker =
conf.getClass(TRACK_IMPL, DefaultStoreFileTracker.class, StoreFileTracker.class);
LOG.info("instantiating StoreFileTracker impl {}", tracker.getName());
@@ -47,22 +49,31 @@ public final class StoreFileTrackerFactory {
}
public static StoreFileTracker create(Configuration conf, boolean isPrimaryReplica, String family,
- HRegionFileSystem regionFs) {
+ HRegionFileSystem regionFs) {
ColumnFamilyDescriptorBuilder fDescBuilder =
ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family));
- StoreContext ctx = StoreContext.getBuilder().
- withColumnFamilyDescriptor(fDescBuilder.build()).
- withRegionFileSystem(regionFs).
- build();
- return StoreFileTrackerFactory.create(conf, isPrimaryReplica, ctx);
+ StoreContext ctx = StoreContext.getBuilder().withColumnFamilyDescriptor(fDescBuilder.build())
+ .withRegionFileSystem(regionFs).build();
+ return StoreFileTrackerFactory.create(conf, TRACK_IMPL, isPrimaryReplica, ctx);
+ }
+
+ public static Configuration mergeConfigurations(Configuration global, TableDescriptor table,
+ ColumnFamilyDescriptor family) {
+ return new CompoundConfiguration().add(global).addBytesMap(table.getValues())
+ .addStringMap(family.getConfiguration()).addBytesMap(family.getValues());
}
- public static Configuration mergeConfigurations(Configuration global,
- TableDescriptor table, ColumnFamilyDescriptor family) {
- return new CompoundConfiguration()
- .add(global)
- .addBytesMap(table.getValues())
- .addStringMap(family.getConfiguration())
- .addBytesMap(family.getValues());
+ static StoreFileTrackerBase create(Configuration conf, String configName,
+ boolean isPrimaryReplica, StoreContext ctx) {
+ String className =
+ Preconditions.checkNotNull(conf.get(configName), "config %s is not set", configName);
+ Class<? extends StoreFileTrackerBase> tracker;
+ try {
+ tracker = Class.forName(className).asSubclass(StoreFileTrackerBase.class);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ LOG.info("instantiating StoreFileTracker impl {} as {}", tracker.getName(), configName);
+ return ReflectionUtils.newInstance(tracker, conf, isPrimaryReplica, ctx);
}
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestMigrationStoreFileTracker.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestMigrationStoreFileTracker.java
new file mode 100644
index 0000000..e9ebb4c
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/storefiletracker/TestMigrationStoreFileTracker.java
@@ -0,0 +1,193 @@
+/**
+ * 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.hadoop.hbase.regionserver.storefiletracker;
+
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.RegionInfoBuilder;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.regionserver.ChunkCreator;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.MemStoreLAB;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.wal.WAL;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.Iterables;
+import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
+
+@RunWith(Parameterized.class)
+@Category({ RegionServerTests.class, MediumTests.class })
+public class TestMigrationStoreFileTracker {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestMigrationStoreFileTracker.class);
+
+ private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
+
+ private static final byte[] CF = Bytes.toBytes("cf");
+
+ private static final byte[] CQ = Bytes.toBytes("cq");
+
+ private static final TableDescriptor TD =
+ TableDescriptorBuilder.newBuilder(TableName.valueOf("file_based_tracker"))
+ .setColumnFamily(ColumnFamilyDescriptorBuilder.of(CF)).build();
+
+ private static final RegionInfo RI = RegionInfoBuilder.newBuilder(TD.getTableName()).build();
+
+ @Rule
+ public TestName name = new TestName();
+
+ @Parameter(0)
+ public Class<? extends StoreFileTrackerBase> srcImplClass;
+
+ @Parameter(1)
+ public Class<? extends StoreFileTrackerBase> dstImplClass;
+
+ private HRegion region;
+
+ private Path rootDir;
+
+ private WAL wal;
+
+ @Parameters(name = "{index}: src={0}, dst={1}")
+ public static List<Object[]> params() {
+ List<Class<? extends StoreFileTrackerBase>> impls =
+ Arrays.asList(DefaultStoreFileTracker.class, FileBasedStoreFileTracker.class);
+ List<Object[]> params = new ArrayList<>();
+ for (Class<? extends StoreFileTrackerBase> src : impls) {
+ for (Class<? extends StoreFileTrackerBase> dst : impls) {
+ if (src.equals(dst)) {
+ continue;
+ }
+ params.add(new Object[] { src, dst });
+ }
+ }
+ return params;
+ }
+
+ @BeforeClass
+ public static void setUpBeforeClass() {
+ ChunkCreator.initialize(MemStoreLAB.CHUNK_SIZE_DEFAULT, false, 0, 0, 0, null,
+ MemStoreLAB.INDEX_CHUNK_SIZE_PERCENTAGE_DEFAULT);
+ }
+
+ @Before
+ public void setUp() throws IOException {
+ Configuration conf = UTIL.getConfiguration();
+ conf.setClass(MigrationStoreFileTracker.SRC_IMPL, srcImplClass, StoreFileTrackerBase.class);
+ conf.setClass(MigrationStoreFileTracker.DST_IMPL, dstImplClass, StoreFileTrackerBase.class);
+ rootDir = UTIL.getDataTestDir(name.getMethodName().replaceAll("[=:\\[ ]", "_"));
+ wal = HBaseTestingUtil.createWal(conf, rootDir, RI);
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ if (region != null) {
+ region.close();
+ }
+ Closeables.close(wal, true);
+ UTIL.cleanupTestDir();
+ }
+
+ private List<String> getStoreFiles() {
+ return Iterables.getOnlyElement(region.getStores()).getStorefiles().stream()
+ .map(s -> s.getFileInfo().getPath().getName()).collect(Collectors.toList());
+ }
+
+ private HRegion createRegion(Class<? extends StoreFileTrackerBase> trackerImplClass)
+ throws IOException {
+ Configuration conf = new Configuration(UTIL.getConfiguration());
+ conf.setClass(StoreFileTrackerFactory.TRACK_IMPL, trackerImplClass, StoreFileTracker.class);
+ return HRegion.createHRegion(RI, rootDir, conf, TD, wal, true);
+ }
+
+ private void reopenRegion(Class<? extends StoreFileTrackerBase> trackerImplClass)
+ throws IOException {
+ region.flush(true);
+ List<String> before = getStoreFiles();
+ region.close();
+ Configuration conf = new Configuration(UTIL.getConfiguration());
+ conf.setClass(StoreFileTrackerFactory.TRACK_IMPL, trackerImplClass, StoreFileTracker.class);
+ region = HRegion.openHRegion(rootDir, RI, TD, wal, conf);
+ List<String> after = getStoreFiles();
+ assertEquals(before.size(), after.size());
+ assertThat(after, hasItems(before.toArray(new String[0])));
+ }
+
+ private void putData(int start, int end) throws IOException {
+ for (int i = start; i < end; i++) {
+ region.put(new Put(Bytes.toBytes(i)).addColumn(CF, CQ, Bytes.toBytes(i)));
+ if (i % 30 == 0) {
+ region.flush(true);
+ }
+ }
+ }
+
+ private void verifyData(int start, int end) throws IOException {
+ for (int i = start; i < end; i++) {
+ Result result = region.get(new Get(Bytes.toBytes(i)));
+ assertEquals(i, Bytes.toInt(result.getValue(CF, CQ)));
+ }
+ }
+
+ @Test
+ public void testMigration() throws IOException {
+ region = createRegion(srcImplClass);
+ putData(0, 100);
+ verifyData(0, 100);
+ reopenRegion(MigrationStoreFileTracker.class);
+ verifyData(0, 100);
+ region.compact(true);
+ putData(100, 200);
+ reopenRegion(dstImplClass);
+ verifyData(0, 200);
+ }
+}