| // Copyright 2017 JanusGraph Authors |
| // |
| // Licensed 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.janusgraph.diskstorage.hbase2; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Joiner; |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.BiMap; |
| import com.google.common.collect.ImmutableBiMap; |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.collect.Sets; |
| import org.apache.hadoop.hbase.HBaseConfiguration; |
| import org.apache.hadoop.hbase.HRegionInfo; |
| import org.apache.hadoop.hbase.HRegionLocation; |
| import org.apache.hadoop.hbase.MasterNotRunningException; |
| import org.apache.hadoop.hbase.ServerName; |
| import org.apache.hadoop.hbase.TableNotEnabledException; |
| import org.apache.hadoop.hbase.TableNotFoundException; |
| import org.apache.hadoop.hbase.ZooKeeperConnectionException; |
| import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor; |
| import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; |
| import org.apache.hadoop.hbase.client.Delete; |
| import org.apache.hadoop.hbase.client.Mutation; |
| import org.apache.hadoop.hbase.client.Put; |
| import org.apache.hadoop.hbase.client.Row; |
| import org.apache.hadoop.hbase.client.TableDescriptor; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.hbase.util.Pair; |
| import org.apache.hadoop.hbase.util.VersionInfo; |
| import org.janusgraph.core.JanusGraphException; |
| import org.janusgraph.diskstorage.BackendException; |
| import org.janusgraph.diskstorage.BaseTransactionConfig; |
| import org.janusgraph.diskstorage.Entry; |
| import org.janusgraph.diskstorage.EntryMetaData; |
| import org.janusgraph.diskstorage.PermanentBackendException; |
| import org.janusgraph.diskstorage.StaticBuffer; |
| import org.janusgraph.diskstorage.StoreMetaData; |
| import org.janusgraph.diskstorage.TemporaryBackendException; |
| import org.janusgraph.diskstorage.common.DistributedStoreManager; |
| import org.janusgraph.diskstorage.configuration.ConfigElement; |
| import org.janusgraph.diskstorage.configuration.ConfigNamespace; |
| import org.janusgraph.diskstorage.configuration.ConfigOption; |
| import org.janusgraph.diskstorage.configuration.Configuration; |
| import org.janusgraph.diskstorage.keycolumnvalue.KCVMutation; |
| import org.janusgraph.diskstorage.keycolumnvalue.KeyColumnValueStore; |
| import org.janusgraph.diskstorage.keycolumnvalue.KeyColumnValueStoreManager; |
| import org.janusgraph.diskstorage.keycolumnvalue.KeyRange; |
| import org.janusgraph.diskstorage.keycolumnvalue.StandardStoreFeatures; |
| import org.janusgraph.diskstorage.keycolumnvalue.StoreFeatures; |
| import org.janusgraph.diskstorage.keycolumnvalue.StoreTransaction; |
| import org.janusgraph.diskstorage.util.BufferUtil; |
| import org.janusgraph.diskstorage.util.StaticArrayBuffer; |
| import org.janusgraph.diskstorage.util.time.TimestampProviders; |
| import org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration; |
| import org.janusgraph.graphdb.configuration.PreInitializeConfigOptions; |
| import org.janusgraph.util.system.IOUtils; |
| import org.janusgraph.util.system.NetworkUtil; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| |
| import static org.janusgraph.diskstorage.Backend.EDGESTORE_NAME; |
| import static org.janusgraph.diskstorage.Backend.INDEXSTORE_NAME; |
| import static org.janusgraph.diskstorage.Backend.LOCK_STORE_SUFFIX; |
| import static org.janusgraph.diskstorage.Backend.SYSTEM_MGMT_LOG_NAME; |
| import static org.janusgraph.diskstorage.Backend.SYSTEM_TX_LOG_NAME; |
| import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.DROP_ON_CLEAR; |
| import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.GRAPH_NAME; |
| import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.IDS_STORE_NAME; |
| import static org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration.SYSTEM_PROPERTIES_STORE_NAME; |
| |
| /** |
| * Storage Manager for HBase |
| */ |
| @PreInitializeConfigOptions |
| public class HBaseStoreManager extends DistributedStoreManager implements KeyColumnValueStoreManager { |
| |
| private static final Logger logger = LoggerFactory.getLogger(HBaseStoreManager.class); |
| |
| public static final ConfigNamespace HBASE_NS = |
| new ConfigNamespace(GraphDatabaseConfiguration.STORAGE_NS, "hbase", "HBase storage options"); |
| |
| public static final ConfigOption<Boolean> SHORT_CF_NAMES = |
| new ConfigOption<>(HBASE_NS, "short-cf-names", |
| "Whether to shorten the names of JanusGraph's column families to one-character mnemonics " + |
| "to conserve storage space", ConfigOption.Type.FIXED, true); |
| |
| public static final String COMPRESSION_DEFAULT = "-DEFAULT-"; |
| |
| public static final ConfigOption<String> COMPRESSION = |
| new ConfigOption<>(HBASE_NS, "compression-algorithm", |
| "An HBase Compression.Algorithm enum string which will be applied to newly created column families. " + |
| "The compression algorithm must be installed and available on the HBase cluster. JanusGraph cannot install " + |
| "and configure new compression algorithms on the HBase cluster by itself.", |
| ConfigOption.Type.MASKABLE, "GZ"); |
| |
| public static final ConfigOption<Boolean> SKIP_SCHEMA_CHECK = |
| new ConfigOption<>(HBASE_NS, "skip-schema-check", |
| "Assume that JanusGraph's HBase table and column families already exist. " + |
| "When this is true, JanusGraph will not check for the existence of its table/CFs, " + |
| "nor will it attempt to create them under any circumstances. This is useful " + |
| "when running JanusGraph without HBase admin privileges.", |
| ConfigOption.Type.MASKABLE, false); |
| |
| public static final ConfigOption<String> HBASE_TABLE = |
| new ConfigOption<>(HBASE_NS, "table", |
| "The name of the table JanusGraph will use. When " + ConfigElement.getPath(SKIP_SCHEMA_CHECK) + |
| " is false, JanusGraph will automatically create this table if it does not already exist." + |
| " If this configuration option is not provided but graph.graphname is, the table will be set" + |
| " to that value.", |
| ConfigOption.Type.LOCAL, "janusgraph"); |
| |
| /** |
| * Related bug fixed in 0.98.0, 0.94.7, 0.95.0: |
| * |
| * https://issues.apache.org/jira/browse/HBASE-8170 |
| */ |
| public static final int MIN_REGION_COUNT = 3; |
| |
| /** |
| * The total number of HBase regions to create with JanusGraph's table. This |
| * setting only effects table creation; this normally happens just once when |
| * JanusGraph connects to an HBase backend for the first time. |
| */ |
| public static final ConfigOption<Integer> REGION_COUNT = |
| new ConfigOption<Integer>(HBASE_NS, "region-count", |
| "The number of initial regions set when creating JanusGraph's HBase table", |
| ConfigOption.Type.MASKABLE, Integer.class, input -> null != input && MIN_REGION_COUNT <= input); |
| |
| /** |
| * This setting is used only when {@link #REGION_COUNT} is unset. |
| * <p/> |
| * If JanusGraph's HBase table does not exist, then it will be created with total |
| * region count = (number of servers reported by ClusterStatus) * (this |
| * value). |
| * <p/> |
| * The Apache HBase manual suggests an order-of-magnitude range of potential |
| * values for this setting: |
| * |
| * <ul> |
| * <li> |
| * <a href="https://hbase.apache.org/book/important_configurations.html#disable.splitting">2.5.2.7. Managed Splitting</a>: |
| * <blockquote> |
| * What's the optimal number of pre-split regions to create? Mileage will |
| * vary depending upon your application. You could start low with 10 |
| * pre-split regions / server and watch as data grows over time. It's |
| * better to err on the side of too little regions and rolling split later. |
| * </blockquote> |
| * </li> |
| * <li> |
| * <a href="https://hbase.apache.org/book/regions.arch.html">9.7 Regions</a>: |
| * <blockquote> |
| * In general, HBase is designed to run with a small (20-200) number of |
| * relatively large (5-20Gb) regions per server... Typically you want to |
| * keep your region count low on HBase for numerous reasons. Usually |
| * right around 100 regions per RegionServer has yielded the best results. |
| * </blockquote> |
| * </li> |
| * </ul> |
| * |
| * These considerations may differ for other HBase implementations (e.g. MapR). |
| */ |
| public static final ConfigOption<Integer> REGIONS_PER_SERVER = |
| new ConfigOption<>(HBASE_NS, "regions-per-server", |
| "The number of regions per regionserver to set when creating JanusGraph's HBase table", |
| ConfigOption.Type.MASKABLE, Integer.class); |
| |
| /** |
| * If this key is present in either the JVM system properties or the process |
| * environment (checked in the listed order, first hit wins), then its value |
| * must be the full package and class name of an implementation of |
| * {@link HBaseCompat} that has a no-arg public constructor. |
| * <p> |
| * When this <b>is not</b> set, JanusGraph attempts to automatically detect the |
| * HBase runtime version by calling {@link VersionInfo#getVersion()}. JanusGraph |
| * then checks the returned version string against a hard-coded list of |
| * supported version prefixes and instantiates the associated compat layer |
| * if a match is found. |
| * <p> |
| * When this <b>is</b> set, JanusGraph will not call |
| * {@code VersionInfo.getVersion()} or read its hard-coded list of supported |
| * version prefixes. JanusGraph will instead attempt to instantiate the class |
| * specified (via the no-arg constructor which must exist) and then attempt |
| * to cast it to HBaseCompat and use it as such. JanusGraph will assume the |
| * supplied implementation is compatible with the runtime HBase version and |
| * make no attempt to verify that assumption. |
| * <p> |
| * Setting this key incorrectly could cause runtime exceptions at best or |
| * silent data corruption at worst. This setting is intended for users |
| * running exotic HBase implementations that don't support VersionInfo or |
| * implementations which return values from {@code VersionInfo.getVersion()} |
| * that are inconsistent with Apache's versioning convention. It may also be |
| * useful to users who want to run against a new release of HBase that JanusGraph |
| * doesn't yet officially support. |
| * |
| */ |
| public static final ConfigOption<String> COMPAT_CLASS = |
| new ConfigOption<>(HBASE_NS, "compat-class", |
| "The package and class name of the HBaseCompat implementation. HBaseCompat masks version-specific HBase API differences. " + |
| "When this option is unset, JanusGraph calls HBase's VersionInfo.getVersion() and loads the matching compat class " + |
| "at runtime. Setting this option forces JanusGraph to instead reflectively load and instantiate the specified class.", |
| ConfigOption.Type.MASKABLE, String.class); |
| |
| public static final int PORT_DEFAULT = 9160; |
| |
| public static final TimestampProviders PREFERRED_TIMESTAMPS = TimestampProviders.MICRO; |
| |
| public static final ConfigNamespace HBASE_CONFIGURATION_NAMESPACE = |
| new ConfigNamespace(HBASE_NS, "ext", "Overrides for hbase-{site,default}.xml options", true); |
| |
| private static final StaticBuffer FOUR_ZERO_BYTES = BufferUtil.zeroBuffer(4); |
| |
| // Immutable instance fields |
| private final BiMap<String, String> shortCfNameMap; |
| private final String tableName; |
| private final String compression; |
| private final int regionCount; |
| private final int regionsPerServer; |
| private final ConnectionMask cnx; |
| private final org.apache.hadoop.conf.Configuration hconf; |
| private final boolean shortCfNames; |
| private final boolean skipSchemaCheck; |
| private final String compatClass; |
| private final HBaseCompat compat; |
| // Cached return value of getDeployment() as requesting it can be expensive. |
| private Deployment deployment = null; |
| |
| private static final ConcurrentHashMap<HBaseStoreManager, Throwable> openManagers = new ConcurrentHashMap<>(); |
| |
| // Mutable instance state |
| private final ConcurrentMap<String, HBaseKeyColumnValueStore> openStores; |
| |
| public HBaseStoreManager(org.janusgraph.diskstorage.configuration.Configuration config) throws BackendException { |
| super(config, PORT_DEFAULT); |
| |
| shortCfNameMap = createShortCfMap(config); |
| |
| Preconditions.checkArgument(null != shortCfNameMap); |
| Collection<String> shorts = shortCfNameMap.values(); |
| Preconditions.checkArgument(Sets.newHashSet(shorts).size() == shorts.size()); |
| |
| checkConfigDeprecation(config); |
| |
| this.tableName = determineTableName(config); |
| this.compression = config.get(COMPRESSION); |
| this.regionCount = config.has(REGION_COUNT) ? config.get(REGION_COUNT) : -1; |
| this.regionsPerServer = config.has(REGIONS_PER_SERVER) ? config.get(REGIONS_PER_SERVER) : -1; |
| this.skipSchemaCheck = config.get(SKIP_SCHEMA_CHECK); |
| this.compatClass = config.has(COMPAT_CLASS) ? config.get(COMPAT_CLASS) : null; |
| this.compat = HBaseCompatLoader.getCompat(compatClass); |
| |
| /* |
| * Specifying both region count options is permitted but may be |
| * indicative of a misunderstanding, so issue a warning. |
| */ |
| if (config.has(REGIONS_PER_SERVER) && config.has(REGION_COUNT)) { |
| logger.warn("Both {} and {} are set in JanusGraph's configuration, but " |
| + "the former takes precedence and the latter will be ignored.", |
| REGION_COUNT, REGIONS_PER_SERVER); |
| } |
| |
| /* This static factory calls HBaseConfiguration.addHbaseResources(), |
| * which in turn applies the contents of hbase-default.xml and then |
| * applies the contents of hbase-site.xml. |
| */ |
| this.hconf = HBaseConfiguration.create(); |
| |
| // Copy a subset of our commons config into a Hadoop config |
| int keysLoaded=0; |
| Map<String,Object> configSub = config.getSubset(HBASE_CONFIGURATION_NAMESPACE); |
| for (Map.Entry<String,Object> entry : configSub.entrySet()) { |
| logger.info("HBase configuration: setting {}={}", entry.getKey(), entry.getValue()); |
| if (entry.getValue()==null) continue; |
| hconf.set(entry.getKey(), entry.getValue().toString()); |
| keysLoaded++; |
| } |
| |
| // Special case for STORAGE_HOSTS |
| if (config.has(GraphDatabaseConfiguration.STORAGE_HOSTS)) { |
| String zkQuorumKey = "hbase.zookeeper.quorum"; |
| String csHostList = Joiner.on(",").join(config.get(GraphDatabaseConfiguration.STORAGE_HOSTS)); |
| hconf.set(zkQuorumKey, csHostList); |
| logger.info("Copied host list from {} to {}: {}", GraphDatabaseConfiguration.STORAGE_HOSTS, zkQuorumKey, csHostList); |
| } |
| |
| logger.debug("HBase configuration: set a total of {} configuration values", keysLoaded); |
| |
| this.shortCfNames = config.get(SHORT_CF_NAMES); |
| |
| try { |
| //this.cnx = HConnectionManager.createConnection(hconf); |
| this.cnx = compat.createConnection(hconf); |
| } catch (IOException e) { |
| throw new PermanentBackendException(e); |
| } |
| |
| if (logger.isTraceEnabled()) { |
| openManagers.put(this, new Throwable("Manager Opened")); |
| dumpOpenManagers(); |
| } |
| |
| logger.debug("Dumping HBase config key=value pairs"); |
| for (Map.Entry<String, String> entry : hconf) { |
| logger.debug("[HBaseConfig] " + entry.getKey() + "=" + entry.getValue()); |
| } |
| logger.debug("End of HBase config key=value pairs"); |
| |
| openStores = new ConcurrentHashMap<>(); |
| } |
| |
| public static BiMap<String, String> createShortCfMap(Configuration config) { |
| return ImmutableBiMap.<String, String>builder() |
| .put(INDEXSTORE_NAME, "g") |
| .put(INDEXSTORE_NAME + LOCK_STORE_SUFFIX, "h") |
| .put(config.get(IDS_STORE_NAME), "i") |
| .put(EDGESTORE_NAME, "e") |
| .put(EDGESTORE_NAME + LOCK_STORE_SUFFIX, "f") |
| .put(SYSTEM_PROPERTIES_STORE_NAME, "s") |
| .put(SYSTEM_PROPERTIES_STORE_NAME + LOCK_STORE_SUFFIX, "t") |
| .put(SYSTEM_MGMT_LOG_NAME, "m") |
| .put(SYSTEM_TX_LOG_NAME, "l") |
| .build(); |
| } |
| |
| @Override |
| public Deployment getDeployment() { |
| if (null != deployment) { |
| return deployment; |
| } |
| |
| List<KeyRange> local; |
| try { |
| local = getLocalKeyPartition(); |
| deployment = null != local && !local.isEmpty() ? Deployment.LOCAL : Deployment.REMOTE; |
| } catch (BackendException e) { |
| throw new RuntimeException(e); |
| } |
| return deployment; |
| } |
| |
| @Override |
| public String toString() { |
| return "hbase[" + tableName + "@" + super.toString() + "]"; |
| } |
| |
| public void dumpOpenManagers() { |
| int estimatedSize = openManagers.size(); |
| logger.trace("---- Begin open HBase store manager list ({} managers) ----", estimatedSize); |
| for (HBaseStoreManager m : openManagers.keySet()) { |
| logger.trace("Manager {} opened at:", m, openManagers.get(m)); |
| } |
| logger.trace("---- End open HBase store manager list ({} managers) ----", estimatedSize); |
| } |
| |
| @Override |
| public void close() { |
| openStores.clear(); |
| if (logger.isTraceEnabled()) |
| openManagers.remove(this); |
| IOUtils.closeQuietly(cnx); |
| } |
| |
| @Override |
| public StoreFeatures getFeatures() { |
| |
| Configuration c = GraphDatabaseConfiguration.buildGraphConfiguration(); |
| |
| StandardStoreFeatures.Builder fb = new StandardStoreFeatures.Builder() |
| .orderedScan(true).unorderedScan(true).batchMutation(true) |
| .multiQuery(true).distributed(true).keyOrdered(true).storeTTL(true) |
| .cellTTL(true).timestamps(true).preferredTimestamps(PREFERRED_TIMESTAMPS) |
| .optimisticLocking(true).keyConsistent(c); |
| |
| try { |
| fb.localKeyPartition(getDeployment() == Deployment.LOCAL); |
| } catch (Exception e) { |
| logger.warn("Unexpected exception during getDeployment()", e); |
| } |
| |
| return fb.build(); |
| } |
| |
| @Override |
| public void mutateMany(Map<String, Map<StaticBuffer, KCVMutation>> mutations, StoreTransaction txh) throws BackendException { |
| final MaskedTimestamp commitTime = new MaskedTimestamp(txh); |
| // In case of an addition and deletion with identical timestamps, the |
| // deletion tombstone wins. |
| // http://hbase.apache.org/book/versions.html#d244e4250 |
| final Map<StaticBuffer, Pair<List<Put>, Delete>> commandsPerKey = |
| convertToCommands( |
| mutations, |
| commitTime.getAdditionTime(times), |
| commitTime.getDeletionTime(times)); |
| |
| final List<Row> batch = new ArrayList<>(commandsPerKey.size()); // actual batch operation |
| |
| // convert sorted commands into representation required for 'batch' operation |
| for (Pair<List<Put>, Delete> commands : commandsPerKey.values()) { |
| if (commands.getFirst() != null && !commands.getFirst().isEmpty()) |
| batch.addAll(commands.getFirst()); |
| |
| if (commands.getSecond() != null) |
| batch.add(commands.getSecond()); |
| } |
| |
| try { |
| TableMask table = null; |
| |
| try { |
| table = cnx.getTable(tableName); |
| table.batch(batch, new Object[batch.size()]); |
| } finally { |
| IOUtils.closeQuietly(table); |
| } |
| } catch (IOException e) { |
| throw new TemporaryBackendException(e); |
| } catch (InterruptedException e) { |
| throw new TemporaryBackendException(e); |
| } |
| |
| this.sleepAfterWrite(commitTime); |
| } |
| |
| @Override |
| public KeyColumnValueStore openDatabase(String longName, StoreMetaData.Container metaData) throws BackendException { |
| // HBase does not support retrieving cell-level TTL by the client. |
| Preconditions.checkArgument(!storageConfig.has(GraphDatabaseConfiguration.STORE_META_TTL, longName) |
| || !storageConfig.get(GraphDatabaseConfiguration.STORE_META_TTL, longName)); |
| |
| HBaseKeyColumnValueStore store = openStores.get(longName); |
| |
| if (store == null) { |
| final String cfName = getCfNameForStoreName(longName); |
| |
| HBaseKeyColumnValueStore newStore = new HBaseKeyColumnValueStore(this, cnx, tableName, cfName, longName); |
| |
| store = openStores.putIfAbsent(longName, newStore); // nothing bad happens if we loose to other thread |
| |
| if (store == null) { |
| if (!skipSchemaCheck) { |
| int cfTTLInSeconds = -1; |
| if (metaData.contains(StoreMetaData.TTL)) { |
| cfTTLInSeconds = metaData.get(StoreMetaData.TTL); |
| } |
| ensureColumnFamilyExists(tableName, cfName, cfTTLInSeconds); |
| } |
| |
| store = newStore; |
| } |
| } |
| |
| return store; |
| } |
| |
| @Override |
| public StoreTransaction beginTransaction(final BaseTransactionConfig config) throws BackendException { |
| return new HBaseTransaction(config); |
| } |
| |
| @Override |
| public String getName() { |
| return tableName; |
| } |
| |
| /** |
| * Deletes the specified table with all its columns. |
| * ATTENTION: Invoking this method will delete the table if it exists and therefore causes data loss. |
| */ |
| @Override |
| public void clearStorage() throws BackendException { |
| try (AdminMask adm = getAdminInterface()) { |
| if (this.storageConfig.get(DROP_ON_CLEAR)) { |
| adm.dropTable(tableName); |
| } else { |
| adm.clearTable(tableName, times.getTime(times.getTime())); |
| } |
| } catch (IOException e) |
| { |
| throw new TemporaryBackendException(e); |
| } |
| } |
| |
| @Override |
| public boolean exists() throws BackendException { |
| try (final AdminMask adm = getAdminInterface()) { |
| return adm.tableExists(tableName); |
| } catch (IOException e) { |
| throw new TemporaryBackendException(e); |
| } |
| } |
| |
| @Override |
| public List<KeyRange> getLocalKeyPartition() throws BackendException { |
| List<KeyRange> result = new LinkedList<>(); |
| try { |
| ensureTableExists( |
| tableName, getCfNameForStoreName(GraphDatabaseConfiguration.SYSTEM_PROPERTIES_STORE_NAME), 0); |
| Map<KeyRange, ServerName> normed = normalizeKeyBounds(cnx.getRegionLocations(tableName)); |
| |
| for (Map.Entry<KeyRange, ServerName> e : normed.entrySet()) { |
| if (NetworkUtil.isLocalConnection(e.getValue().getHostname())) { |
| result.add(e.getKey()); |
| logger.debug("Found local key/row partition {} on host {}", e.getKey(), e.getValue()); |
| } else { |
| logger.debug("Discarding remote {}", e.getValue()); |
| } |
| } |
| } catch (MasterNotRunningException e) { |
| logger.warn("Unexpected MasterNotRunningException", e); |
| } catch (ZooKeeperConnectionException e) { |
| logger.warn("Unexpected ZooKeeperConnectionException", e); |
| } catch (IOException e) { |
| logger.warn("Unexpected IOException", e); |
| } |
| return result; |
| } |
| |
| /** |
| * each key from an {@link HRegionInfo} to a {@link KeyRange} expressing the |
| * region's start and end key bounds using JanusGraph-partitioning-friendly |
| * conventions (start inclusive, end exclusive, zero bytes appended where |
| * necessary to make all keys at least 4 bytes long). |
| * <p/> |
| * This method iterates over the entries in its map parameter and performs |
| * the following conditional conversions on its keys. "Require" below means |
| * either a {@link Preconditions} invocation or an assertion. HRegionInfo |
| * sometimes returns start and end keys of zero length; this method replaces |
| * zero length keys with null before doing any of the checks described |
| * below. The parameter map and the values it contains are only read and |
| * never modified. |
| * |
| * <ul> |
| * <li>If an entry's HRegionInfo has null start and end keys, then first |
| * require that the parameter map is a singleton, and then return a |
| * single-entry map whose {@code KeyRange} has start and end buffers that |
| * are both four bytes of zeros.</li> |
| * <li>If the entry has a null end key (but non-null start key), put an |
| * equivalent entry in the result map with a start key identical to the |
| * input, except that zeros are appended to values less than 4 bytes long, |
| * and an end key that is four bytes of zeros. |
| * <li>If the entry has a null start key (but non-null end key), put an |
| * equivalent entry in the result map where the start key is four bytes of |
| * zeros, and the end key has zeros appended, if necessary, to make it at |
| * least 4 bytes long, after which one is added to the padded value in |
| * unsigned 32-bit arithmetic with overflow allowed.</li> |
| * <li>Any entry which matches none of the above criteria results in an |
| * equivalent entry in the returned map, except that zeros are appended to |
| * both keys to make each at least 4 bytes long, and the end key is then |
| * incremented as described in the last bullet point.</li> |
| * </ul> |
| * |
| * After iterating over the parameter map, this method checks that it either |
| * saw no entries with null keys, one entry with a null start key and a |
| * different entry with a null end key, or one entry with both start and end |
| * keys null. If any null keys are observed besides these three cases, the |
| * method will die with a precondition failure. |
| * |
| * @param locations A list of HRegionInfo |
| * @return JanusGraph-friendly expression of each region's rowkey boundaries |
| */ |
| private Map<KeyRange, ServerName> normalizeKeyBounds(List<HRegionLocation> locations) { |
| |
| HRegionLocation nullStart = null; |
| HRegionLocation nullEnd = null; |
| |
| ImmutableMap.Builder<KeyRange, ServerName> b = ImmutableMap.builder(); |
| |
| for (HRegionLocation location : locations) { |
| HRegionInfo regionInfo = location.getRegionInfo(); |
| ServerName serverName = location.getServerName(); |
| byte startKey[] = regionInfo.getStartKey(); |
| byte endKey[] = regionInfo.getEndKey(); |
| |
| if (0 == startKey.length) { |
| startKey = null; |
| logger.trace("Converted zero-length HBase startKey byte array to null"); |
| } |
| |
| if (0 == endKey.length) { |
| endKey = null; |
| logger.trace("Converted zero-length HBase endKey byte array to null"); |
| } |
| |
| if (null == startKey && null == endKey) { |
| Preconditions.checkState(1 == locations.size()); |
| logger.debug("HBase table {} has a single region {}", tableName, regionInfo); |
| // Choose arbitrary shared value = startKey = endKey |
| return b.put(new KeyRange(FOUR_ZERO_BYTES, FOUR_ZERO_BYTES), serverName).build(); |
| } else if (null == startKey) { |
| logger.debug("Found HRegionInfo with null startKey on server {}: {}", serverName, regionInfo); |
| Preconditions.checkState(null == nullStart); |
| nullStart = location; |
| // I thought endBuf would be inclusive from the HBase javadoc, but in practice it is exclusive |
| StaticBuffer endBuf = StaticArrayBuffer.of(zeroExtend(endKey)); |
| // Replace null start key with zeroes |
| b.put(new KeyRange(FOUR_ZERO_BYTES, endBuf), serverName); |
| } else if (null == endKey) { |
| logger.debug("Found HRegionInfo with null endKey on server {}: {}", serverName, regionInfo); |
| Preconditions.checkState(null == nullEnd); |
| nullEnd = location; |
| // Replace null end key with zeroes |
| b.put(new KeyRange(StaticArrayBuffer.of(zeroExtend(startKey)), FOUR_ZERO_BYTES), serverName); |
| } else { |
| Preconditions.checkState(null != startKey); |
| Preconditions.checkState(null != endKey); |
| |
| // Convert HBase's inclusive end keys into exclusive JanusGraph end keys |
| StaticBuffer startBuf = StaticArrayBuffer.of(zeroExtend(startKey)); |
| StaticBuffer endBuf = StaticArrayBuffer.of(zeroExtend(endKey)); |
| |
| KeyRange kr = new KeyRange(startBuf, endBuf); |
| b.put(kr, serverName); |
| logger.debug("Found HRegionInfo with non-null end and start keys on server {}: {}", serverName, regionInfo); |
| } |
| } |
| |
| // Require either no null key bounds or a pair of them |
| Preconditions.checkState(!(null == nullStart ^ null == nullEnd)); |
| |
| // Check that every key in the result is at least 4 bytes long |
| Map<KeyRange, ServerName> result = b.build(); |
| for (KeyRange kr : result.keySet()) { |
| Preconditions.checkState(4 <= kr.getStart().length()); |
| Preconditions.checkState(4 <= kr.getEnd().length()); |
| } |
| |
| return result; |
| } |
| |
| /** |
| * If the parameter is shorter than 4 bytes, then create and return a new 4 |
| * byte array with the input array's bytes followed by zero bytes. Otherwise |
| * return the parameter. |
| * |
| * @param dataToPad non-null but possibly zero-length byte array |
| * @return either the parameter or a new array |
| */ |
| private final byte[] zeroExtend(byte[] dataToPad) { |
| assert null != dataToPad; |
| |
| final int targetLength = 4; |
| |
| if (targetLength <= dataToPad.length) |
| return dataToPad; |
| |
| byte padded[] = new byte[targetLength]; |
| |
| for (int i = 0; i < dataToPad.length; i++) |
| padded[i] = dataToPad[i]; |
| |
| for (int i = dataToPad.length; i < padded.length; i++) |
| padded[i] = (byte)0; |
| |
| return padded; |
| } |
| |
| public static String shortenCfName(BiMap<String, String> shortCfNameMap, String longName) throws PermanentBackendException { |
| final String s; |
| if (shortCfNameMap.containsKey(longName)) { |
| s = shortCfNameMap.get(longName); |
| Preconditions.checkNotNull(s); |
| logger.debug("Substituted default CF name \"{}\" with short form \"{}\" to reduce HBase KeyValue size", longName, s); |
| } else { |
| if (shortCfNameMap.containsValue(longName)) { |
| String fmt = "Must use CF long-form name \"%s\" instead of the short-form name \"%s\" when configured with %s=true"; |
| String msg = String.format(fmt, shortCfNameMap.inverse().get(longName), longName, SHORT_CF_NAMES.getName()); |
| throw new PermanentBackendException(msg); |
| } |
| s = longName; |
| logger.debug("Kept default CF name \"{}\" because it has no associated short form", s); |
| } |
| return s; |
| } |
| |
| private TableDescriptor ensureTableExists(String tableName, String initialCFName, int ttlInSeconds) throws BackendException { |
| AdminMask adm = null; |
| |
| TableDescriptor desc; |
| |
| try { // Create our table, if necessary |
| adm = getAdminInterface(); |
| /* |
| * Some HBase versions/impls respond badly to attempts to create a |
| * table without at least one CF. See #661. Creating a CF along with |
| * the table avoids HBase carping. |
| */ |
| if (adm.tableExists(tableName)) { |
| desc = adm.getTableDescriptor(tableName); |
| // Check and warn if long and short cf names are mixedly used for the same table. |
| if (shortCfNames && initialCFName.equals(shortCfNameMap.get(SYSTEM_PROPERTIES_STORE_NAME))) { |
| String longCFName = shortCfNameMap.inverse().get(initialCFName); |
| if (desc.getColumnFamily(Bytes.toBytes(longCFName)) != null) { |
| logger.warn("Configuration {}=true, but the table \"{}\" already has column family with long name \"{}\".", |
| SHORT_CF_NAMES.getName(), tableName, longCFName); |
| logger.warn("Check {} configuration.", SHORT_CF_NAMES.getName()); |
| } |
| } |
| else if (!shortCfNames && initialCFName.equals(SYSTEM_PROPERTIES_STORE_NAME)) { |
| String shortCFName = shortCfNameMap.get(initialCFName); |
| if (desc.getColumnFamily(Bytes.toBytes(shortCFName)) != null) { |
| logger.warn("Configuration {}=false, but the table \"{}\" already has column family with short name \"{}\".", |
| SHORT_CF_NAMES.getName(), tableName, shortCFName); |
| logger.warn("Check {} configuration.", SHORT_CF_NAMES.getName()); |
| } |
| } |
| } else { |
| desc = createTable(tableName, initialCFName, ttlInSeconds, adm); |
| } |
| } catch (IOException e) { |
| throw new TemporaryBackendException(e); |
| } finally { |
| IOUtils.closeQuietly(adm); |
| } |
| |
| return desc; |
| } |
| |
| private TableDescriptor createTable(String tableName, String cfName, int ttlInSeconds, AdminMask adm) throws IOException { |
| TableDescriptor desc = compat.newTableDescriptor(tableName); |
| |
| ColumnFamilyDescriptor cdesc = ColumnFamilyDescriptorBuilder.of(cfName); |
| cdesc = setCFOptions(cdesc, ttlInSeconds); |
| |
| desc = compat.addColumnFamilyToTableDescriptor(desc, cdesc); |
| |
| int count; // total regions to create |
| String src; |
| |
| if (MIN_REGION_COUNT <= (count = regionCount)) { |
| src = "region count configuration"; |
| } else if (0 < regionsPerServer && |
| MIN_REGION_COUNT <= (count = regionsPerServer * adm.getEstimatedRegionServerCount())) { |
| src = "ClusterStatus server count"; |
| } else { |
| count = -1; |
| src = "default"; |
| } |
| |
| if (MIN_REGION_COUNT < count) { |
| adm.createTable(desc, getStartKey(count), getEndKey(count), count); |
| logger.debug("Created table {} with region count {} from {}", tableName, count, src); |
| } else { |
| adm.createTable(desc); |
| logger.debug("Created table {} with default start key, end key, and region count", tableName); |
| } |
| |
| return desc; |
| } |
| |
| /** |
| * <p/> |
| * From the {@code createTable} javadoc: |
| * "The start key specified will become the end key of the first region of |
| * the table, and the end key specified will become the start key of the |
| * last region of the table (the first region has a null start key and |
| * the last region has a null end key)" |
| * <p/> |
| * To summarize, the {@code createTable} argument called "startKey" is |
| * actually the end key of the first region. |
| */ |
| private byte[] getStartKey(int regionCount) { |
| ByteBuffer regionWidth = ByteBuffer.allocate(4); |
| regionWidth.putInt((int)(((1L << 32) - 1L) / regionCount)).flip(); |
| return StaticArrayBuffer.of(regionWidth).getBytes(0, 4); |
| } |
| |
| /** |
| * Companion to {@link #getStartKey(int)}. See its javadoc for details. |
| */ |
| private byte[] getEndKey(int regionCount) { |
| ByteBuffer regionWidth = ByteBuffer.allocate(4); |
| regionWidth.putInt((int)(((1L << 32) - 1L) / regionCount * (regionCount - 1))).flip(); |
| return StaticArrayBuffer.of(regionWidth).getBytes(0, 4); |
| } |
| |
| private void ensureColumnFamilyExists(String tableName, String columnFamily, int ttlInSeconds) throws BackendException { |
| AdminMask adm = null; |
| try { |
| adm = getAdminInterface(); |
| TableDescriptor desc = ensureTableExists(tableName, columnFamily, ttlInSeconds); |
| |
| Preconditions.checkNotNull(desc); |
| |
| ColumnFamilyDescriptor cf = desc.getColumnFamily(Bytes.toBytes(columnFamily)); |
| |
| // Create our column family, if necessary |
| if (cf == null) { |
| try { |
| if (!adm.isTableDisabled(tableName)) { |
| adm.disableTable(tableName); |
| } |
| } catch (TableNotEnabledException e) { |
| logger.debug("Table {} already disabled", tableName); |
| } catch (IOException e) { |
| throw new TemporaryBackendException(e); |
| } |
| |
| try { |
| ColumnFamilyDescriptor cdesc = ColumnFamilyDescriptorBuilder.of(columnFamily); |
| |
| setCFOptions(cdesc, ttlInSeconds); |
| |
| adm.addColumn(tableName, cdesc); |
| |
| try { |
| logger.debug("Added HBase ColumnFamily {}, waiting for 1 sec. to propogate.", columnFamily); |
| Thread.sleep(1000L); |
| } catch (InterruptedException ie) { |
| throw new TemporaryBackendException(ie); |
| } |
| |
| adm.enableTable(tableName); |
| } catch (TableNotFoundException ee) { |
| logger.error("TableNotFoundException", ee); |
| throw new PermanentBackendException(ee); |
| } catch (org.apache.hadoop.hbase.TableExistsException ee) { |
| logger.debug("Swallowing exception {}", ee); |
| } catch (IOException ee) { |
| throw new TemporaryBackendException(ee); |
| } |
| } |
| } finally { |
| IOUtils.closeQuietly(adm); |
| } |
| } |
| |
| private ColumnFamilyDescriptor setCFOptions(ColumnFamilyDescriptor cdesc, int ttlInSeconds) { |
| ColumnFamilyDescriptor ret = null; |
| |
| if (null != compression && !compression.equals(COMPRESSION_DEFAULT)) { |
| ret = compat.setCompression(cdesc, compression); |
| } |
| |
| if (ttlInSeconds > 0) { |
| ret = ColumnFamilyDescriptorBuilder.newBuilder(cdesc).setTimeToLive(ttlInSeconds).build(); |
| } |
| |
| return ret; |
| } |
| |
| /** |
| * Convert JanusGraph internal Mutation representation into HBase native commands. |
| * |
| * @param mutations Mutations to convert into HBase commands. |
| * @param putTimestamp The timestamp to use for Put commands. |
| * @param delTimestamp The timestamp to use for Delete commands. |
| * @return Commands sorted by key converted from JanusGraph internal representation. |
| * @throws org.janusgraph.diskstorage.PermanentBackendException |
| */ |
| @VisibleForTesting |
| Map<StaticBuffer, Pair<List<Put>, Delete>> convertToCommands(Map<String, Map<StaticBuffer, KCVMutation>> mutations, |
| final long putTimestamp, |
| final long delTimestamp) throws PermanentBackendException { |
| // A map of rowkey to commands (list of Puts, Delete) |
| final Map<StaticBuffer, Pair<List<Put>, Delete>> commandsPerKey = new HashMap<>(); |
| |
| for (Map.Entry<String, Map<StaticBuffer, KCVMutation>> entry : mutations.entrySet()) { |
| |
| String cfString = getCfNameForStoreName(entry.getKey()); |
| byte[] cfName = Bytes.toBytes(cfString); |
| |
| for (Map.Entry<StaticBuffer, KCVMutation> m : entry.getValue().entrySet()) { |
| final byte[] key = m.getKey().as(StaticBuffer.ARRAY_FACTORY); |
| KCVMutation mutation = m.getValue(); |
| |
| Pair<List<Put>, Delete> commands = commandsPerKey.get(m.getKey()); |
| |
| // The firt time we go through the list of input <rowkey, KCVMutation>, |
| // create the holder for a particular rowkey |
| if (commands == null) { |
| commands = new Pair<>(); |
| // List of all the Puts for this rowkey, including the ones without TTL and with TTL. |
| final List<Put> putList = new ArrayList<>(); |
| commands.setFirst(putList); |
| commandsPerKey.put(m.getKey(), commands); |
| } |
| |
| if (mutation.hasDeletions()) { |
| if (commands.getSecond() == null) { |
| Delete d = new Delete(key); |
| compat.setTimestamp(d, delTimestamp); |
| commands.setSecond(d); |
| } |
| |
| for (StaticBuffer b : mutation.getDeletions()) { |
| // commands.getSecond() is a Delete for this rowkey. |
| commands.getSecond().addColumns(cfName, b.as(StaticBuffer.ARRAY_FACTORY), delTimestamp); |
| } |
| } |
| |
| if (mutation.hasAdditions()) { |
| // All the entries (column cells) with the rowkey use this one Put, except the ones with TTL. |
| final Put putColumnsWithoutTtl = new Put(key, putTimestamp); |
| // At the end of this loop, there will be one Put entry in the commands.getFirst() list that |
| // contains all additions without TTL set, and possible multiple Put entries for columns |
| // that have TTL set. |
| for (Entry e : mutation.getAdditions()) { |
| |
| // Deal with TTL within the entry (column cell) first |
| // HBase cell level TTL is actually set at the Mutation/Put level. |
| // Therefore we need to construct a new Put for each entry (column cell) with TTL. |
| // We can not combine them because column cells within the same rowkey may: |
| // 1. have no TTL |
| // 2. have TTL |
| // 3. have different TTL |
| final Integer ttl = (Integer) e.getMetaData().get(EntryMetaData.TTL); |
| if (null != ttl && ttl > 0) { |
| // Create a new Put |
| Put putColumnWithTtl = new Put(key, putTimestamp); |
| addColumnToPut(putColumnWithTtl, cfName, putTimestamp, e); |
| // Convert ttl from second (JanusGraph TTL) to millisec (HBase TTL) |
| // @see JanusGraphManagement#setTTL(JanusGraphSchemaType, Duration) |
| // Cast Put to Mutation for backward compatibility with HBase 0.98.x |
| // HBase supports cell-level TTL for versions 0.98.6 and above. |
| ((Mutation) putColumnWithTtl).setTTL(ttl * 1000); |
| // commands.getFirst() is the list of Puts for this rowkey. Add this |
| // Put column with TTL to the list. |
| commands.getFirst().add(putColumnWithTtl); |
| } else { |
| addColumnToPut(putColumnsWithoutTtl, cfName, putTimestamp, e); |
| } |
| } |
| // If there were any mutations without TTL set, add them to commands.getFirst() |
| if (!putColumnsWithoutTtl.isEmpty()) { |
| commands.getFirst().add(putColumnsWithoutTtl); |
| } |
| } |
| } |
| } |
| |
| return commandsPerKey; |
| } |
| |
| private void addColumnToPut(Put p, byte[] cfName, long putTimestamp, Entry e) { |
| p.addColumn(cfName, e.getColumnAs(StaticBuffer.ARRAY_FACTORY), putTimestamp, |
| e.getValueAs(StaticBuffer.ARRAY_FACTORY)); |
| } |
| |
| private String getCfNameForStoreName(String storeName) throws PermanentBackendException { |
| return shortCfNames ? shortenCfName(shortCfNameMap, storeName) : storeName; |
| } |
| |
| private void checkConfigDeprecation(org.janusgraph.diskstorage.configuration.Configuration config) { |
| if (config.has(GraphDatabaseConfiguration.STORAGE_PORT)) { |
| logger.warn("The configuration property {} is ignored for HBase. Set hbase.zookeeper.property.clientPort in hbase-site.xml or {}.hbase.zookeeper.property.clientPort in JanusGraph's configuration file.", |
| ConfigElement.getPath(GraphDatabaseConfiguration.STORAGE_PORT), ConfigElement.getPath(HBASE_CONFIGURATION_NAMESPACE)); |
| } |
| } |
| |
| private AdminMask getAdminInterface() { |
| try { |
| return cnx.getAdmin(); |
| } catch (IOException e) { |
| throw new JanusGraphException(e); |
| } |
| } |
| |
| private String determineTableName(org.janusgraph.diskstorage.configuration.Configuration config) { |
| if ((!config.has(HBASE_TABLE)) && (config.has(GRAPH_NAME))) { |
| return config.get(GRAPH_NAME); |
| } |
| return config.get(HBASE_TABLE); |
| } |
| } |
