Google Git
Sign in
apache / jackrabbit-oak / b0c185f0e59d320dba0782704366f3c6031f8b9f / . / oak-store-document / src / main / java / org / apache / jackrabbit / oak / plugins / document / rdb / RDBDocumentStore.java
blob: 15676af7a77ab783ca0710f441e3ab31797f8da3 [file] [log] [blame]
/*
* 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.jackrabbit.oak.plugins.document.rdb;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.Lists.newArrayList;
import static com.google.common.collect.Lists.partition;
import static org.apache.jackrabbit.oak.plugins.document.UpdateUtils.checkConditions;
import static org.apache.jackrabbit.oak.plugins.document.rdb.RDBJDBCTools.asDocumentStoreException;
import static org.apache.jackrabbit.oak.plugins.document.rdb.RDBJDBCTools.closeResultSet;
import static org.apache.jackrabbit.oak.plugins.document.rdb.RDBJDBCTools.closeStatement;
import static org.apache.jackrabbit.oak.plugins.document.rdb.RDBJDBCTools.createTableName;
import static org.apache.jackrabbit.oak.plugins.document.util.Utils.getModuleVersion;
import java.io.ByteArrayOutputStream;
import java.io.Closeable;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.zip.Deflater;
import java.util.zip.GZIPOutputStream;
import javax.sql.DataSource;
import org.apache.jackrabbit.oak.cache.CacheStats;
import org.apache.jackrabbit.oak.cache.CacheValue;
import org.apache.jackrabbit.oak.plugins.document.Collection;
import org.apache.jackrabbit.oak.plugins.document.Document;
import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStoreBuilder;
import org.apache.jackrabbit.oak.plugins.document.DocumentStore;
import org.apache.jackrabbit.oak.plugins.document.DocumentStoreException;
import org.apache.jackrabbit.oak.plugins.document.DocumentStoreStatsCollector;
import org.apache.jackrabbit.oak.plugins.document.NodeDocument;
import org.apache.jackrabbit.oak.plugins.document.UpdateOp;
import org.apache.jackrabbit.oak.plugins.document.UpdateOp.Key;
import org.apache.jackrabbit.oak.plugins.document.UpdateOp.Operation;
import org.apache.jackrabbit.oak.plugins.document.UpdateUtils;
import org.apache.jackrabbit.oak.plugins.document.cache.CacheChangesTracker;
import org.apache.jackrabbit.oak.plugins.document.cache.CacheInvalidationStats;
import org.apache.jackrabbit.oak.plugins.document.cache.ModificationStamp;
import org.apache.jackrabbit.oak.plugins.document.cache.NodeDocumentCache;
import org.apache.jackrabbit.oak.plugins.document.locks.NodeDocumentLocks;
import org.apache.jackrabbit.oak.plugins.document.locks.StripedNodeDocumentLocks;
import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore;
import org.apache.jackrabbit.oak.plugins.document.util.CloseableIterator;
import org.apache.jackrabbit.oak.plugins.document.util.SystemPropertySupplier;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Function;
import com.google.common.base.Stopwatch;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
/**
* Implementation of {@link DocumentStore} for relational databases.
*
* <h3 id="apidocs.supported-databases">Supported Databases</h3>
* <p>
* The code is supposed to be sufficiently generic to run with a variety of
* database implementations. However, the tables are created when required to
* simplify testing, and <em>that</em> code specifically supports these
* databases:
* <ul>
* <li>H2DB</li>
* <li>Apache Derby</li>
* <li>IBM DB2</li>
* <li>PostgreSQL</li>
* <li>MariaDB (MySQL)</li>
* <li>Microsoft SQL Server</li>
* <li>Oracle</li>
* </ul>
*
* <h3 id="apidocs.table-layout">Table Layout</h3>
* <p>
* Data for each of the DocumentStore's {@link Collection}s is stored in its own
* database table (with a name matching the collection).
* <p>
* The tables essentially implement key/value storage, where the key usually is
* derived from an Oak path, and the value is a serialization of a
* {@link Document} (or a part of one). Additional fields are used for queries,
* debugging, and concurrency control:
* <table style="text-align: left;">
* <caption></caption>
* <thead>
* <tr>
* <th>Column</th>
* <th>Type</th>
* <th>Description</th>
* </tr>
* </thead> <tbody>
* <tr>
* <th>ID</th>
* <td>varchar(512) not null primary key</td>
* <td>The document's key (for databases that can not handle 512 character
* primary keys, such as MySQL, varbinary is possible as well).</td>
* </tr>
* <tr>
* <th>MODIFIED</th>
* <td>bigint</td>
* <td>Low-resolution timestamp.
* </tr>
* <tr>
* <th>HASBINARY</th>
* <td>smallint</td>
* <td>Flag indicating whether the document has binary properties.
* </tr>
* <tr>
* <th>DELETEDONCE</th>
* <td>smallint</td>
* <td>Flag indicating whether the document has been deleted once.
* </tr>
* <tr>
* <th>MODCOUNT</th>
* <td>bigint</td>
* <td>Modification counter, used for avoiding overlapping updates.</td>
* </tr>
* <tr>
* <th>DSIZE</th>
* <td>bigint</td>
* <td>The approximate size of the document's JSON serialization (for debugging
* purposes).</td>
* </tr>
* <tr>
* <th>VERSION</th>
* <td>smallint</td>
* <td>The schema version the code writing to a row (or inserting it) was aware
* of (introduced with schema version 1). Not set for rows written by version 0
* client code.</td>
* </tr>
* <tr>
* <th>SDTYPE</th>
* <td>smallint</td>
* <td>Split Document type.</td>
* </tr>
* <tr>
* <th>SDMAXREVTIME</th>
* <td>bigint</td>
* <td>Split document max revision time..</td>
* </tr>
* <tr>
* <th>DATA</th>
* <td>varchar(16384)</td>
* <td>The document's JSON serialization (only used for small document sizes, in
* which case BDATA (below) is not set), or a sequence of JSON serialized update
* operations to be applied against the last full serialization.</td>
* </tr>
* <tr>
* <th>BDATA</th>
* <td>blob</td>
* <td>The document's JSON serialization (usually GZIPped, only used for "large"
* documents).</td>
* </tr>
* </tbody>
* </table>
* <p>
* The names of database tables can be prefixed; the purpose is mainly for
* testing, as tables can also be dropped automatically when the store is
* disposed (this only happens for those tables that have been created on
* demand).
* <h4 id="apidocs.versioning">Versioning</h4>
* <p>
* The initial database layout used in OAK 1.0 through 1.6 is version 0.
* <p>
* Version 1 introduces an additional "version" column, which records the schema
* version of the code writing to the database (upon insert and update). This is
* in preparation of future layout changes which might introduce new columns.
* <p>
* Version 2 introduces an additional "sdtype" and "sdmaxrevtime".
* <p>
* The code deals with both version 0, version 1 and version 2 table layouts. By
* default, it tries to create version 2 tables, and also tries to upgrade
* existing version 0 and 1 tables to version 2.
* <h4>DB-specific information</h4>
* <p>
* Databases need to be configured so that:
* <ul>
* <li>Text fields support all Unicode code points,</li>
* <li>Collation of text fields happens by Unicode code point,</li>
* <li>and BLOBs need to support at least 16 MB.</li>
* </ul>
* <p>
* See the
* <a href="https://jackrabbit.apache.org/oak/docs/nodestore/document/rdb-document-store.html#database-creation">RDBDocumentStore documentation</a>
* for more information.
* <h3 id="apidocs.table-creation">Table Creation</h3>
* <p>
* The code tries to create the tables when they are not present. Likewise, it
* tries to upgrade to a newer schema when needed.
* <p>
* Users/Administrators who prefer to stay in control over table generation can
* create them "manually". The oak-run "<a href="https://jackrabbit.apache.org/oak/docs/nodestore/document/rdb-document-store.html#rdbddldump"><code>rdbddldump</code></a>"
* command can be used to print out the DDL statements that would have been used for auto-creation
* and/or automatic schema updates.
*
* <h3>Caching</h3>
* <p>
* The cache borrows heavily from the {@link MongoDocumentStore} implementation.
*
* <h3>Queries</h3>
* <p>
* The implementation currently supports only three indexed properties: "_bin",
* "deletedOnce", and "_modified". Attempts to use a different indexed property
* will cause a {@link DocumentStoreException}.
*/
public class RDBDocumentStore implements DocumentStore {
/**
* Creates a {@linkplain RDBDocumentStore} instance using the provided
* {@link DataSource}, {@link DocumentNodeStoreBuilder}, and {@link RDBOptions}.
*/
public RDBDocumentStore(DataSource ds, DocumentNodeStoreBuilder<?> builder, RDBOptions options) {
try {
initialize(ds, builder, options);
} catch (Exception ex) {
throw asDocumentStoreException(ex, "initializing RDB document store");
}
}
/**
* Creates a {@linkplain RDBDocumentStore} instance using the provided
* {@link DataSource}, {@link DocumentNodeStoreBuilder}, and default
* {@link RDBOptions}.
*/
public RDBDocumentStore(DataSource ds, DocumentNodeStoreBuilder<?> builder) {
this(ds, builder, new RDBOptions());
}
@Override
public <T extends Document> T find(Collection<T> collection, String id) {
return find(collection, id, Integer.MAX_VALUE);
}
@Override
public <T extends Document> T find(final Collection<T> collection, final String id, int maxCacheAge) {
return readDocumentCached(collection, id, maxCacheAge);
}
@NotNull
@Override
public <T extends Document> List<T> query(Collection<T> collection, String fromKey, String toKey, int limit) {
return query(collection, fromKey, toKey, null, 0, limit);
}
@NotNull
@Override
public <T extends Document> List<T> query(Collection<T> collection, String fromKey, String toKey, String indexedProperty,
long startValue, int limit) {
List<QueryCondition> conditions = Collections.emptyList();
if (indexedProperty != null) {
conditions = Collections.singletonList(new QueryCondition(indexedProperty, ">=", startValue));
}
return internalQuery(collection, fromKey, toKey, EMPTY_KEY_PATTERN, conditions, limit);
}
@NotNull
protected <T extends Document> List<T> query(Collection<T> collection, String fromKey, String toKey,
List<String> excludeKeyPatterns, List<QueryCondition> conditions, int limit) {
return internalQuery(collection, fromKey, toKey, excludeKeyPatterns, conditions, limit);
}
@Override
public <T extends Document> void remove(Collection<T> collection, String id) {
try {
delete(collection, id);
} finally {
invalidateCache(collection, id, true);
}
}
@Override
public <T extends Document> void remove(Collection<T> collection, List<String> ids) {
try {
delete(collection, ids);
} finally {
for (String id : ids) {
invalidateCache(collection, id, true);
}
}
}
@Override
public <T extends Document> int remove(Collection<T> collection, Map<String, Long> toRemove) {
try {
return delete(collection, toRemove);
} finally {
for (String id : toRemove.keySet()) {
invalidateCache(collection, id, true);
}
}
}
@Override
public <T extends Document> int remove(Collection<T> collection, String indexedProperty, long startValue, long endValue)
throws DocumentStoreException {
try {
List<QueryCondition> conditions = new ArrayList<QueryCondition>();
conditions.add(new QueryCondition(indexedProperty, ">", startValue));
conditions.add(new QueryCondition(indexedProperty, "<", endValue));
return deleteWithCondition(collection, conditions);
} finally {
if (collection == Collection.NODES) {
// this method is currently being used only for Journal
// collection while GC. But, to keep sanctity of the API, we
// need to acknowledge that Nodes collection could've been used.
// But, in this signature, there's no useful way to invalidate
// cache.
// So, we use the hammer for this task
invalidateCache();
}
}
}
@Override
public <T extends Document> boolean create(Collection<T> collection, List<UpdateOp> updateOps) {
return internalCreate(collection, updateOps);
}
@Override
public <T extends Document> T createOrUpdate(Collection<T> collection, UpdateOp update) {
UpdateUtils.assertUnconditional(update);
return internalCreateOrUpdate(collection, update, update.isNew(), false, RETRIES);
}
@Override
public <T extends Document> List<T> createOrUpdate(Collection<T> collection, List<UpdateOp> updateOps) {
// fall back to sequential mode if batches are turned off using system
// property, or the number of update operations is small
if (!BATCHUPDATES || updateOps.size() < MINIMALBULKUPDATESIZE) {
List<T> results = new ArrayList<T>(updateOps.size());
for (UpdateOp update : updateOps) {
results.add(createOrUpdate(collection, update));
}
return results;
} else {
return internalCreateOrUpdate(collection, updateOps);
}
}
private static int MINIMALBULKUPDATESIZE = 3;
private <T extends Document> List<T> internalCreateOrUpdate(Collection<T> collection, List<UpdateOp> updateOps) {
final Stopwatch watch = startWatch();
Map<UpdateOp, T> results = new LinkedHashMap<UpdateOp, T>();
Map<String, UpdateOp> operationsToCover = new LinkedHashMap<String, UpdateOp>();
Set<UpdateOp> duplicates = new HashSet<UpdateOp>();
for (UpdateOp updateOp : updateOps) {
UpdateUtils.assertUnconditional(updateOp);
if (operationsToCover.containsKey(updateOp.getId())) {
duplicates.add(updateOp);
results.put(updateOp, null);
} else {
UpdateOp clone = updateOp.copy();
addUpdateCounters(clone);
operationsToCover.put(clone.getId(), clone);
results.put(clone, null);
}
}
Map<String, T> oldDocs = new HashMap<String, T>();
if (collection == Collection.NODES) {
oldDocs.putAll(readDocumentCached(collection, operationsToCover.keySet()));
}
int i = 0; // iteration count
while (operationsToCover.size() >= MINIMALBULKUPDATESIZE) {
// We should try to insert documents only during the first
// iteration. In the 2nd and 3rd iterations we only deal with
// conflicting documents, so they already exist in the database
// and there's no point in inserting them.
boolean upsert = i == 0;
if (i++ == 3) {
// operations that conflicted in 3 consecutive bulk
// updates should be applied sequentially
break;
}
for (List<UpdateOp> partition : partition(newArrayList(operationsToCover.values()), CHUNKSIZE)) {
Map<UpdateOp, T> successfulUpdates = bulkUpdate(collection, partition, oldDocs, upsert);
results.putAll(successfulUpdates);
operationsToCover.values().removeAll(successfulUpdates.keySet());
}
}
// if there are some changes left, we'll apply them one after another
for (UpdateOp updateOp : updateOps) {
UpdateOp conflictedOp = operationsToCover.remove(updateOp.getId());
if (conflictedOp != null) {
if (collection == Collection.NODES) {
LOG.debug("createOrUpdate: update conflict on {}, invalidating cache and retrying...", updateOp.getId());
nodesCache.invalidate(updateOp.getId());
} else {
LOG.debug("createOrUpdate: update conflict on {}, retrying...", updateOp.getId());
}
results.put(conflictedOp, createOrUpdate(collection, updateOp));
} else if (duplicates.contains(updateOp)) {
results.put(updateOp, createOrUpdate(collection, updateOp));
}
}
stats.doneCreateOrUpdate(watch.elapsed(TimeUnit.NANOSECONDS),
collection, Lists.transform(updateOps, new Function<UpdateOp, String>() {
@Override
public String apply(UpdateOp input) {
return input.getId();
}
}));
return new ArrayList<T>(results.values());
}
private <T extends Document> Map<String, T> readDocumentCached(Collection<T> collection, Set<String> keys) {
Map<String, T> documents = new HashMap<String, T>();
if (collection == Collection.NODES) {
for (String key : keys) {
NodeDocument cached = nodesCache.getIfPresent(key);
if (cached != null && cached != NodeDocument.NULL) {
T doc = castAsT(unwrap(cached));
documents.put(doc.getId(), doc);
}
}
}
Set<String> documentsToRead = Sets.difference(keys, documents.keySet());
Map<String, T> readDocuments = readDocumentsUncached(collection, documentsToRead);
documents.putAll(readDocuments);
if (collection == Collection.NODES) {
for (T doc : readDocuments.values()) {
nodesCache.putIfAbsent((NodeDocument) doc);
}
}
return documents;
}
private <T extends Document> Map<String, T> readDocumentsUncached(Collection<T> collection, Set<String> keys) {
Map<String, T> result = new HashMap<String, T>();
Connection connection = null;
RDBTableMetaData tmd = getTable(collection);
try {
connection = this.ch.getROConnection();
List<RDBRow> rows = db.read(connection, tmd, keys);
int size = rows.size();
for (int i = 0; i < size; i++) {
RDBRow row = rows.set(i, null);
T document = convertFromDBObject(collection, row);
result.put(document.getId(), document);
}
connection.commit();
} catch (Exception ex) {
throw asDocumentStoreException(ex, "trying to read: " + keys);
} finally {
this.ch.closeConnection(connection);
}
return result;
}
@Nullable
private <T extends Document> CacheChangesTracker obtainTracker(Collection<T> collection, Set<String> keys) {
if (collection == Collection.NODES) {
return this.nodesCache.registerTracker(keys);
} else {
return null;
}
}
@Nullable
private <T extends Document> CacheChangesTracker obtainTracker(Collection<T> collection, String fromKey, String toKey) {
if (collection == Collection.NODES) {
return this.nodesCache.registerTracker(fromKey, toKey);
} else {
return null;
}
}
private <T extends Document> Map<UpdateOp, T> bulkUpdate(Collection<T> collection, List<UpdateOp> updates, Map<String, T> oldDocs, boolean upsert) {
Set<String> missingDocs = new HashSet<String>();
for (UpdateOp op : updates) {
if (!oldDocs.containsKey(op.getId())) {
missingDocs.add(op.getId());
}
}
if (LOG.isTraceEnabled()) {
LOG.trace("bulkUpdate: cached docs to be updated: {}", dumpKeysAndModcounts(oldDocs));
}
Map<String, T> freshDocs = readDocumentsUncached(collection, missingDocs);
if (LOG.isTraceEnabled()) {
LOG.trace("bulkUpdate: fresh docs to be updated: {}", dumpKeysAndModcounts(freshDocs));
}
oldDocs.putAll(freshDocs);
try (CacheChangesTracker tracker = obtainTracker(collection, Sets.union(oldDocs.keySet(), missingDocs) )) {
List<T> docsToUpdate = new ArrayList<T>(updates.size());
Set<String> keysToUpdate = new HashSet<String>();
for (UpdateOp update : updates) {
String id = update.getId();
T modifiedDoc = collection.newDocument(this);
T oldDoc = oldDocs.get(id);
if (oldDoc != null) {
oldDoc.deepCopy(modifiedDoc);
}
UpdateUtils.applyChanges(modifiedDoc, update);
if (oldDoc != null || update.isNew()) {
// only create if updateOp allows it
docsToUpdate.add(modifiedDoc);
}
keysToUpdate.add(id);
}
Connection connection = null;
RDBTableMetaData tmd = getTable(collection);
try {
connection = this.ch.getRWConnection();
Set<String> successfulUpdates = db.update(connection, tmd, docsToUpdate, upsert);
connection.commit();
Set<String> failedUpdates = Sets.difference(keysToUpdate, successfulUpdates);
oldDocs.keySet().removeAll(failedUpdates);
if (LOG.isTraceEnabled()) {
LOG.trace("bulkUpdate: success for {}, failure for {}", successfulUpdates, failedUpdates);
}
if (collection == Collection.NODES) {
List<NodeDocument> docsToCache = new ArrayList<>();
for (T doc : docsToUpdate) {
if (successfulUpdates.contains(doc.getId())) {
docsToCache.add((NodeDocument) doc);
}
}
nodesCache.putNonConflictingDocs(tracker, docsToCache);
}
Map<UpdateOp, T> result = new HashMap<UpdateOp, T>();
for (UpdateOp op : updates) {
if (successfulUpdates.contains(op.getId())) {
result.put(op, oldDocs.get(op.getId()));
}
}
return result;
} catch (SQLException ex) {
this.ch.rollbackConnection(connection);
throw handleException("update failed for: " + keysToUpdate, ex, collection, keysToUpdate);
} finally {
this.ch.closeConnection(connection);
}
}
}
@Override
public <T extends Document> T findAndUpdate(Collection<T> collection, UpdateOp update) {
return internalCreateOrUpdate(collection, update, false, true, RETRIES);
}
@Override
public CacheInvalidationStats invalidateCache() {
for (CacheValue key : nodesCache.keys()) {
invalidateCache(Collection.NODES, key.toString());
}
return null;
}
@Override
public CacheInvalidationStats invalidateCache(Iterable<String> keys) {
for (String key : keys) {
invalidateCache(Collection.NODES, key);
}
return null;
}
@Override
public <T extends Document> void invalidateCache(Collection<T> collection, String id) {
invalidateCache(collection, id, false);
}
private <T extends Document> void invalidateCache(Collection<T> collection, String id, boolean remove) {
if (collection == Collection.NODES) {
invalidateNodesCache(id, remove);
}
}
private void invalidateNodesCache(String id, boolean remove) {
try (CacheLock lock = acquireLockFor(id)) {
if (remove) {
nodesCache.invalidate(id);
} else {
nodesCache.markChanged(id);
NodeDocument entry = nodesCache.getIfPresent(id);
if (entry != null) {
entry.markUpToDate(0);
}
}
}
}
@Override
public long determineServerTimeDifferenceMillis() {
Connection connection = null;
try {
connection = this.ch.getROConnection();
long result = this.db.determineServerTimeDifferenceMillis(connection);
connection.commit();
return result;
} catch (SQLException ex) {
LOG.error("Trying to determine time difference to server", ex);
throw asDocumentStoreException(ex, "Trying to determine time difference to server");
} finally {
this.ch.closeConnection(connection);
}
}
// used for diagnostics
private String droppedTables = "";
public String getDroppedTables() {
return this.droppedTables;
}
// table names
private static Map<Collection<? extends Document>, String> TABLEMAP;
private static List<String> TABLENAMES;
static {
Map<Collection<? extends Document>, String> tmp = new HashMap<Collection<? extends Document>, String>();
tmp.put(Collection.CLUSTER_NODES, "CLUSTERNODES");
tmp.put(Collection.JOURNAL, "JOURNAL");
tmp.put(Collection.NODES, "NODES");
tmp.put(Collection.SETTINGS, "SETTINGS");
TABLEMAP = Collections.unmodifiableMap(tmp);
List<String> tl = new ArrayList<String>(TABLEMAP.values());
Collections.sort(tl);
TABLENAMES = Collections.unmodifiableList(tl);
}
public static List<String> getTableNames() {
return TABLENAMES;
}
/**
* Holds the data about a table that can vary: name, whether the primary key
* is binary, and the estimated size of the "data" column.
*/
static class RDBTableMetaData {
private final String catalog;
private final String name;
private boolean idIsBinary = false;
private boolean dataIsNChar = false;
private boolean hasVersion = false;
private boolean hasSplitDocs = false;
private int dataLimitInOctets = 16384;
private String schemaInfo = "";
private String indexInfo = "";
private Set<String> columnOnlyProperties = Collections.unmodifiableSet(COLUMNPROPERTIES);
private Set<String> columnProperties = Collections.unmodifiableSet(COLUMNPROPERTIES);
public RDBTableMetaData(@Nullable String catalog, @NotNull String name) {
this.catalog = catalog == null ? "" : catalog;
this.name = name;
}
public int getDataLimitInOctets() {
return this.dataLimitInOctets;
}
public String getCatalog() {
return this.catalog;
}
public Set<String> getColumnProperties() {
return this.columnProperties;
}
public Set<String> getColumnOnlyProperties() {
return this.columnOnlyProperties;
}
public String getIndexInfo() {
return this.indexInfo;
}
public String getName() {
return this.name;
}
public String getSchemaInfo() {
return this.schemaInfo;
}
public boolean isDataNChar() {
return this.dataIsNChar;
}
public boolean isIdBinary() {
return this.idIsBinary;
}
public boolean hasSplitDocs() {
return this.hasSplitDocs;
}
public boolean hasVersion() {
return this.hasVersion;
}
public void setDataIsNChar(boolean dataIsNChar) {
this.dataIsNChar = dataIsNChar;
}
public void setIdIsBinary(boolean idIsBinary) {
this.idIsBinary = idIsBinary;
}
public void setHasSplitDocs(boolean hasSplitDocs) {
this.hasSplitDocs = hasSplitDocs;
this.columnProperties = Collections.unmodifiableSet(hasSplitDocs ? COLUMNPROPERTIES2 : COLUMNPROPERTIES) ;
}
public void setHasVersion(boolean hasVersion) {
this.hasVersion = hasVersion;
}
public void setDataLimitInOctets(int dataLimitInOctets) {
this.dataLimitInOctets = dataLimitInOctets;
}
public void setSchemaInfo(String schemaInfo) {
this.schemaInfo = schemaInfo;
}
public void setIndexInfo(String indexInfo) {
this.indexInfo = indexInfo;
}
}
private final Map<Collection<? extends Document>, RDBTableMetaData> tableMeta = new HashMap<Collection<? extends Document>, RDBTableMetaData>();
@Override
public void dispose() {
if (!this.tablesToBeDropped.isEmpty()) {
String dropped = "";
LOG.debug("attempting to drop: " + this.tablesToBeDropped);
for (String tname : this.tablesToBeDropped) {
Connection con = null;
try {
con = this.ch.getRWConnection();
Statement stmt = null;
try {
stmt = con.createStatement();
stmt.execute("drop table " + tname);
stmt.close();
con.commit();
dropped += tname + " ";
} catch (SQLException ex) {
LOG.debug("attempting to drop: " + tname, ex);
} finally {
closeStatement(stmt);
}
} catch (SQLException ex) {
LOG.debug("attempting to drop: " + tname, ex);
} finally {
this.ch.closeConnection(con);
}
}
this.droppedTables = dropped.trim();
}
try {
this.ch.close();
} catch (IOException ex) {
LOG.error("closing connection handler", ex);
}
try {
this.nodesCache.close();
} catch (IOException ex) {
LOG.warn("Error occurred while closing nodes cache", ex);
}
LOG.info("RDBDocumentStore (" + getModuleVersion() + ") disposed" + getCnStats()
+ (this.droppedTables.isEmpty() ? "" : " (tables dropped: " + this.droppedTables + ")"));
}
@Override
public <T extends Document> T getIfCached(Collection<T> collection, String id) {
if (collection != Collection.NODES) {
return null;
} else {
NodeDocument doc = nodesCache.getIfPresent(id);
doc = (doc != null) ? unwrap(doc) : null;
return castAsT(doc);
}
}
private <T extends Document> T getIfCached(Collection<T> collection, String id, long modCount) {
T doc = getIfCached(collection, id);
if (doc != null && doc.getModCount() != null && doc.getModCount() == modCount) {
return doc;
} else {
return null;
}
}
@Override
public Iterable<CacheStats> getCacheStats() {
return nodesCache.getCacheStats();
}
@Override
public Map<String, String> getMetadata() {
return metadata;
}
/**
* Statistics are generated for each table. The following fields are always
* added:
* <dl>
* <dt><em>tableName</em>.ns</dt>
* <dd>fully qualified name of the database table</dd>
* <dt><em>tableName</em>.schemaInfo</dt>
* <dd>DDL information for table, as obtained during startup</dd>
* <dt><em>tableName</em>.indexInfo</dt>
* <dd>DDL information for associated indexes, as obtained during
* startup</dd>
* <dt><em>tableName</em>.count</dt>
* <dd>exact number of rows</dd>
* </dl>
* In addition, some statistics information for
* {@link Collection#CLUSTER_NODES} is added:
* <dl>
* <dt>clusterNodes.updates</dt>
* <dd>Writes to the table, counted by cluster node ID</dd>
* </dl>
* Finally, additional database-specific statistics may be added; see
* descriptions in
* {@link RDBDocumentStoreDB#getAdditionalStatistics(RDBConnectionHandler, String, String)}
* for details.
**/
@NotNull
@Override
public Map<String, String> getStats() {
ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
tableMeta.forEach((k, v) -> toMapBuilder(builder, k, v));
if (LOG.isDebugEnabled()) {
LOG.debug("statistics obtained: " + builder.toString());
}
return builder.build();
}
private <T extends Document> void toMapBuilder(ImmutableMap.Builder<String, String> builder, Collection<T> collection, RDBTableMetaData meta) {
String prefix = collection.toString();
builder.put(prefix + ".ns", meta.getCatalog() + "." + meta.getName());
builder.put(prefix + ".schemaInfo", meta.getSchemaInfo());
builder.put(prefix + ".indexInfo", meta.getIndexInfo());
if (Collection.CLUSTER_NODES.equals(collection)) {
builder.put(prefix + ".updates", getCnStats());
}
// live data
Map<String, String> map = this.dbInfo.getAdditionalStatistics(this.ch, meta.getCatalog(), meta.getName());
map.forEach((k, v) -> builder.put(prefix + "." + k, v));
try {
long c = queryCount(collection, null, null, Collections.emptyList(), Collections.emptyList());
builder.put(prefix + ".count", Long.toString(c));
}
catch (DocumentStoreException ex) {
LOG.debug("getting entry count for " + prefix, ex);
}
}
// implementation
private static final String MODIFIED = "_modified";
private static final String MODCOUNT = "_modCount";
/**
* Optional counter for changes to "_collisions" map ({@link NodeDocument#COLLISIONS}).
*/
public static final String COLLISIONSMODCOUNT = "_collisionsModCount";
private static final String ID = "_id";
private static final Logger LOG = LoggerFactory.getLogger(RDBDocumentStore.class);
private Exception callStack;
private RDBConnectionHandler ch;
// from options
private Set<String> tablesToBeDropped = new HashSet<String>();
// ratio between Java characters and UTF-8 encoding
// a) single characters will fit into 3 bytes
// b) a surrogate pair (two Java characters) will fit into 4 bytes
// thus...
public static final int CHAR2OCTETRATIO = 3;
// number of retries for updates
private static final int RETRIES = 10;
// see OAK-2044
protected static final boolean USECMODCOUNT = true;
// Database schema supported by this version
protected static final int SCHEMA = 2;
private static final Key MODIFIEDKEY = new Key(MODIFIED, null);
// DB-specific information
private RDBDocumentStoreDB dbInfo;
// utility class for performing low-level operations
private RDBDocumentStoreJDBC db;
protected static final List<String> EMPTY_KEY_PATTERN = Collections.emptyList();
private Map<String, String> metadata;
private DocumentStoreStatsCollector stats;
private boolean readOnly;
// VERSION column mapping in queries used by RDBVersionGCSupport
public static String VERSIONPROP = "__version";
// set of supported indexed properties
private static final Set<String> INDEXEDPROPERTIES = new HashSet<String>(Arrays.asList(new String[] { MODIFIED,
NodeDocument.HAS_BINARY_FLAG, NodeDocument.DELETED_ONCE, NodeDocument.SD_TYPE, NodeDocument.SD_MAX_REV_TIME_IN_SECS, VERSIONPROP }));
// set of required table columns
private static final Set<String> REQUIREDCOLUMNS = Collections.unmodifiableSet(new HashSet<String>(Arrays.asList(
new String[] { "id", "dsize", "deletedonce", "bdata", "data", "cmodcount", "modcount", "hasbinary", "modified" })));
// set of optional table columns
private static final Set<String> OPTIONALCOLUMNS = Collections
.unmodifiableSet(new HashSet<String>(Arrays.asList(new String[] { "version", "sdtype", "sdmaxrevtime" })));
// set of properties not serialized to JSON
private static final Set<String> COLUMNPROPERTIES = new HashSet<String>(Arrays.asList(
new String[] { ID, NodeDocument.HAS_BINARY_FLAG, NodeDocument.DELETED_ONCE, COLLISIONSMODCOUNT, MODIFIED, MODCOUNT }));
// set of properties not serialized to JSON, schema version 2
private static final Set<String> COLUMNPROPERTIES2 = new HashSet<String>(Arrays.asList(
new String[] { ID, NodeDocument.HAS_BINARY_FLAG, NodeDocument.DELETED_ONCE, COLLISIONSMODCOUNT, MODIFIED, MODCOUNT,
NodeDocument.SD_TYPE, NodeDocument.SD_MAX_REV_TIME_IN_SECS, VERSIONPROP }));
private final RDBDocumentSerializer ser = new RDBDocumentSerializer(this);
private void initialize(DataSource ds, DocumentNodeStoreBuilder<?> builder, RDBOptions options) throws Exception {
this.stats = builder.getDocumentStoreStatsCollector();
this.callStack = LOG.isDebugEnabled() ? new Exception("call stack of RDBDocumentStore creation") : null;
this.readOnly = builder.getReadOnlyMode();
this.ch = new RDBConnectionHandler(ds);
Connection con = this.ch.getRWConnection();
String catalog = con.getCatalog();
DatabaseMetaData md = con.getMetaData();
if (null == catalog) {
// Oracle
catalog = md.getUserName();
}
this.tableMeta.put(Collection.NODES,
new RDBTableMetaData(catalog, createTableName(options.getTablePrefix(), TABLEMAP.get(Collection.NODES))));
this.tableMeta.put(Collection.CLUSTER_NODES,
new RDBTableMetaData(catalog, createTableName(options.getTablePrefix(), TABLEMAP.get(Collection.CLUSTER_NODES))));
this.tableMeta.put(Collection.JOURNAL,
new RDBTableMetaData(catalog, createTableName(options.getTablePrefix(), TABLEMAP.get(Collection.JOURNAL))));
this.tableMeta.put(Collection.SETTINGS,
new RDBTableMetaData(catalog, createTableName(options.getTablePrefix(), TABLEMAP.get(Collection.SETTINGS))));
this.locks = new StripedNodeDocumentLocks();
this.nodesCache = builder.buildNodeDocumentCache(this, locks);
int isolation = con.getTransactionIsolation();
String isolationDiags = RDBJDBCTools.isolationLevelToString(isolation);
if (isolation != Connection.TRANSACTION_READ_COMMITTED) {
LOG.info("Detected transaction isolation level " + isolationDiags + " is "
+ (isolation < Connection.TRANSACTION_READ_COMMITTED ? "lower" : "higher") + " than expected "
+ RDBJDBCTools.isolationLevelToString(Connection.TRANSACTION_READ_COMMITTED)
+ " - check datasource configuration");
}
String dbDesc = String.format("%s %s (%d.%d)", md.getDatabaseProductName(), md.getDatabaseProductVersion(),
md.getDatabaseMajorVersion(), md.getDatabaseMinorVersion()).replaceAll("[\r\n\t]", " ").trim();
String driverDesc = String.format("%s %s (%d.%d)", md.getDriverName(), md.getDriverVersion(), md.getDriverMajorVersion(),
md.getDriverMinorVersion()).replaceAll("[\r\n\t]", " ").trim();
String dbUrl = md.getURL();
this.dbInfo = RDBDocumentStoreDB.getValue(md.getDatabaseProductName());
this.db = new RDBDocumentStoreJDBC(this.dbInfo, this.ser, QUERYHITSLIMIT, QUERYTIMELIMIT);
this.metadata = ImmutableMap.<String,String>builder()
.put("type", "rdb")
.put("db", md.getDatabaseProductName())
.put("version", md.getDatabaseProductVersion())
.put("driver", md.getDriverName())
.put("driverVersion", md.getDriverVersion())
.build();
String versionDiags = dbInfo.checkVersion(md);
if (!versionDiags.isEmpty()) {
LOG.error(versionDiags);
}
if (! "".equals(dbInfo.getInitializationStatement())) {
Statement stmt = null;
try {
stmt = con.createStatement();
stmt.execute(dbInfo.getInitializationStatement());
stmt.close();
con.commit();
}
finally {
closeStatement(stmt);
}
}
List<String> tablesCreated = new ArrayList<String>();
List<String> tablesPresent = new ArrayList<String>();
try {
createTableFor(con, Collection.CLUSTER_NODES, this.tableMeta.get(Collection.CLUSTER_NODES), tablesCreated,
tablesPresent, options.getInitialSchema(), options.getUpgradeToSchema());
createTableFor(con, Collection.NODES, this.tableMeta.get(Collection.NODES), tablesCreated, tablesPresent,
options.getInitialSchema(), options.getUpgradeToSchema());
createTableFor(con, Collection.SETTINGS, this.tableMeta.get(Collection.SETTINGS), tablesCreated, tablesPresent,
options.getInitialSchema(), options.getUpgradeToSchema());
createTableFor(con, Collection.JOURNAL, this.tableMeta.get(Collection.JOURNAL), tablesCreated, tablesPresent,
options.getInitialSchema(), options.getUpgradeToSchema());
} finally {
con.commit();
con.close();
}
StringBuilder tableDiags = new StringBuilder();
RDBTableMetaData nodesMeta = this.tableMeta.get(Collection.NODES);
tableDiags.append(nodesMeta.getSchemaInfo());
if (!nodesMeta.getIndexInfo().isEmpty()) {
tableDiags.append(" /* ").append(nodesMeta.getIndexInfo()).append(" */");
}
if (options.isDropTablesOnClose()) {
tablesToBeDropped.addAll(tablesCreated);
}
if (tableDiags.length() != 0) {
tableDiags.insert(0, ", ");
}
Map<String, String> diag = dbInfo.getAdditionalDiagnostics(this.ch, this.tableMeta.get(Collection.NODES).getName());
LOG.info("RDBDocumentStore (" + getModuleVersion() + ") instantiated for database " + dbDesc + ", using driver: "
+ driverDesc + ", connecting to: " + dbUrl + (diag.isEmpty() ? "" : (", properties: " + diag.toString()))
+ ", transaction isolation level: " + isolationDiags + tableDiags);
if (!tablesPresent.isEmpty()) {
LOG.info("Tables present upon startup: " + tablesPresent);
}
if (!tablesCreated.isEmpty()) {
LOG.info("Tables created upon startup: " + tablesCreated
+ (options.isDropTablesOnClose() ? " (will be dropped on exit)" : ""));
}
}
private static boolean isBinaryType(int sqlType) {
return sqlType == Types.VARBINARY || sqlType == Types.BINARY || sqlType == Types.LONGVARBINARY;
}
private static boolean isNChar(int sqlType) {
return sqlType == Types.NCHAR || sqlType == Types.NVARCHAR || sqlType == Types.LONGNVARCHAR;
}
private static void obtainFlagsFromResultSetMeta(ResultSetMetaData met, RDBTableMetaData tmd) throws SQLException {
for (int i = 1; i <= met.getColumnCount(); i++) {
String lcName = met.getColumnName(i).toLowerCase(Locale.ENGLISH);
if ("id".equals(lcName)) {
tmd.setIdIsBinary(isBinaryType(met.getColumnType(i)));
}
if ("data".equals(lcName)) {
tmd.setDataLimitInOctets(met.getPrecision(i));
tmd.setDataIsNChar(isNChar(met.getColumnType(i)));
}
if ("version".equals(lcName)) {
tmd.setHasVersion(true);
}
if ("sdtype".equals(lcName)) {
tmd.setHasSplitDocs(true);
}
}
}
private static String asQualifiedDbName(String one, String two) {
one = Strings.nullToEmpty(one).trim();
two = Strings.nullToEmpty(two).trim();
if (one.isEmpty() && two.isEmpty()) {
return null;
} else {
one = Strings.nullToEmpty(one).trim();
two = two == null ? "" : two.trim();
return one.isEmpty() ? two : one + "." + two;
}
}
private static String indexTypeAsString(int type) {
switch (type) {
case DatabaseMetaData.tableIndexClustered:
return "clustered";
case DatabaseMetaData.tableIndexHashed:
return "hashed";
case DatabaseMetaData.tableIndexStatistic:
return "statistic";
case DatabaseMetaData.tableIndexOther:
return "other";
default:
return "indexType=" + type;
}
}
private static String dumpIndexData(DatabaseMetaData met, ResultSetMetaData rmet, String tableName, Set<String> indexedColumns) {
ResultSet rs = null;
try {
// if the result set metadata provides a table name, use that (the
// other one might be inaccurate due to case insensitivity issues)
String rmetTableName = Strings.nullToEmpty(rmet.getTableName(1)).trim();
if (!rmetTableName.isEmpty()) {
tableName = rmetTableName;
}
String rmetSchemaName = Strings.nullToEmpty(rmet.getSchemaName(1)).trim();
rs = met.getIndexInfo(null, null, tableName, false, true);
Map<String, IndexInformation> indices = getIndexInformation(rs, rmetSchemaName);
if (indices.isEmpty() && !tableName.equals(tableName.toUpperCase(Locale.ENGLISH))) {
// might have failed due to the DB's handling on ucase/lcase,
// retry ucase
rs = met.getIndexInfo(null, null, tableName.toUpperCase(Locale.ENGLISH), false, true);
indices = getIndexInformation(rs, rmetSchemaName);
}
if (indexedColumns != null) {
for (IndexInformation idata : indices.values()) {
indexedColumns.addAll(idata.columns);
}
}
return dumpIndexData(indices);
} catch (SQLException ex) {
// well it was best-effort
String message = String.format("exception while retrieving index information: %s, code %d, state %s", ex.getMessage(),
ex.getErrorCode(), ex.getSQLState());
LOG.debug(message, ex);
return "/* " + message + "*/";
} finally {
closeResultSet(rs);
}
}
private static String dumpIndexData(Map<String, IndexInformation> indices) {
StringBuilder sb = new StringBuilder();
for (Entry<String, IndexInformation> index : indices.entrySet()) {
String indexName = index.getKey();
IndexInformation info = index.getValue();
if (!info.fields.isEmpty()) {
if (sb.length() != 0) {
sb.append(", ");
}
sb.append(String.format("%sindex %s on %s (", info.nonunique ? "" : "unique ", indexName, info.tname));
String delim = "";
for (String field : info.fields.values()) {
sb.append(delim);
delim = ", ";
sb.append(field);
}
sb.append(")");
sb.append(" ").append(info.type);
}
if (info.filterCondition != null) {
sb.append(" where ").append(info.filterCondition);
}
sb.append(String.format(" (#%d, p%d)", info.cardinality, info.pages));
}
return sb.toString();
}
// see https://docs.oracle.com/javase/7/docs/api/java/sql/DatabaseMetaData.html#getIndexInfo(java.lang.String,%20java.lang.String,%20java.lang.String,%20boolean,%20boolean)
private static Map<String, IndexInformation> getIndexInformation(ResultSet rs, String rmetSchemaName) throws SQLException {
Map<String, IndexInformation> result = new TreeMap<>();
while (rs.next()) {
String name = asQualifiedDbName(rs.getString("INDEX_QUALIFIER"), rs.getString("INDEX_NAME"));
if (name != null) {
IndexInformation info = result.get(name);
if (info == null) {
info = new IndexInformation();
result.put(name, info);
info.fields = new TreeMap<>();
}
info.nonunique = rs.getBoolean("NON_UNIQUE");
info.type = indexTypeAsString(rs.getInt("TYPE"));
String inSchema = rs.getString("TABLE_SCHEM");
inSchema = Strings.nullToEmpty(inSchema).trim();
String filterCondition = Strings.nullToEmpty(rs.getString("FILTER_CONDITION")).trim();
if (!filterCondition.isEmpty()) {
info.filterCondition = filterCondition;
}
info.cardinality = rs.getInt("CARDINALITY");
info.pages = rs.getInt("PAGES");
Set<String> columns = new HashSet<>();
info.columns = columns;
// skip indices on tables in other schemas in case we have that information
if (rmetSchemaName.isEmpty() || inSchema.isEmpty() || rmetSchemaName.equals(inSchema)) {
String tname = asQualifiedDbName(inSchema, rs.getString("TABLE_NAME"));
info.tname = tname;
String cname = rs.getString("COLUMN_NAME");
if (cname != null) {
columns.add(cname.toUpperCase(Locale.ENGLISH));
String order = "A".equals(rs.getString("ASC_OR_DESC")) ? " ASC" : ("D".equals(rs.getString("ASC_OR_DESC")) ? " DESC" : "");
info.fields.put(rs.getInt("ORDINAL_POSITION"), cname + order);
}
}
}
}
return result;
}
private static class IndexInformation {
public Map<Integer, String> fields;
public boolean nonunique;
public String type;
public String filterCondition;
public int cardinality;
public int pages;
public Set<String> columns;
public String tname;
}
private void createTableFor(Connection con, Collection<? extends Document> col, RDBTableMetaData tmd, List<String> tablesCreated,
List<String> tablesPresent, int initialSchema, int upgradeToSchema) throws SQLException {
String dbname = this.dbInfo.toString();
if (con.getMetaData().getURL() != null) {
dbname += " (" + con.getMetaData().getURL() + ")";
}
String tableName = tmd.getName();
Statement checkStatement = null;
ResultSet checkResultSet = null;
Statement creatStatement = null;
try {
// avoid PreparedStatement due to weird DB2 behavior (OAK-6237)
checkStatement = con.createStatement();
checkResultSet = checkStatement.executeQuery("select * from " + tableName + " where ID = '0'");
// try to discover size of DATA column and binary-ness of ID
ResultSetMetaData met = checkResultSet.getMetaData();
obtainFlagsFromResultSetMeta(met, tmd);
// check that all required columns are present
Set<String> requiredColumns = new HashSet<String>(REQUIREDCOLUMNS);
Set<String> unknownColumns = new HashSet<String>();
boolean hasVersionColumn = false;
boolean hasSDTypeColumn = false;
for (int i = 1; i <= met.getColumnCount(); i++) {
String cname = met.getColumnName(i).toLowerCase(Locale.ENGLISH);
if (!requiredColumns.remove(cname)) {
if (!OPTIONALCOLUMNS.contains(cname)) {
unknownColumns.add(cname);
}
}
if (cname.equals("version")) {
hasVersionColumn = true;
}
if (cname.equals("sdtype")) {
hasSDTypeColumn = true;
}
}
if (!requiredColumns.isEmpty()) {
String message = String.format("Table %s: the following required columns are missing: %s", tableName,
requiredColumns.toString());
LOG.error(message);
throw new DocumentStoreException(message);
}
if (!unknownColumns.isEmpty()) {
String message = String.format("Table %s: the following columns are unknown and will not be maintained: %s",
tableName, unknownColumns.toString());
LOG.info(message);
}
String tableInfo = RDBJDBCTools.dumpResultSetMeta(met);
tmd.setSchemaInfo(tableInfo);
Set<String> indexOn = new HashSet<String>();
String indexInfo = dumpIndexData(con.getMetaData(), met, tableName, indexOn);
tmd.setIndexInfo(indexInfo);
closeResultSet(checkResultSet);
boolean dbWasChanged = false;
if (this.readOnly) {
LOG.debug("Skipping table update code because store is initialized in readOnly mode");
}
else {
if (!hasVersionColumn && upgradeToSchema >= 1) {
dbWasChanged |= upgradeTable(con, tableName, 1);
}
if (!hasSDTypeColumn && upgradeToSchema >= 2) {
dbWasChanged |= upgradeTable(con, tableName, 2);
}
if (!indexOn.contains("MODIFIED") && col == Collection.NODES) {
dbWasChanged |= addModifiedIndex(con, tableName);
}
}
tablesPresent.add(tableName);
if (dbWasChanged) {
getTableMetaData(con, col, tmd);
}
} catch (SQLException ex) {
// table does not appear to exist
con.rollback();
LOG.debug("trying to read from '" + tableName + "'", ex);
if (this.readOnly) {
throw new SQLException("Would like to create table '" + tableName
+ "', but RDBDocumentStore has been initialized in 'readonly' mode");
}
try {
creatStatement = con.createStatement();
creatStatement.execute(this.dbInfo.getTableCreationStatement(tableName, initialSchema));
creatStatement.close();
for (String ic : this.dbInfo.getIndexCreationStatements(tableName, initialSchema)) {
creatStatement = con.createStatement();
creatStatement.execute(ic);
creatStatement.close();
}
con.commit();
if (initialSchema < 1 && upgradeToSchema >= 1) {
upgradeTable(con, tableName, 1);
}
if (initialSchema < 2 && upgradeToSchema >= 2) {
upgradeTable(con, tableName, 2);
}
tablesCreated.add(tableName);
getTableMetaData(con, col, tmd);
}
catch (SQLException ex2) {
LOG.error("Failed to create table '" + tableName + "' in '" + dbname + "'", ex2);
throw ex2;
}
}
finally {
closeResultSet(checkResultSet);
closeStatement(checkStatement);
closeStatement(creatStatement);
}
}
private boolean upgradeTable(Connection con, String tableName, int level) throws SQLException {
boolean wasChanged = false;
for (String statement : this.dbInfo.getTableUpgradeStatements(tableName, level)) {
Statement upgradeStatement = null;
try {
upgradeStatement = con.createStatement();
upgradeStatement.execute(statement);
upgradeStatement.close();
con.commit();
LOG.info("Upgraded " + tableName + " to DB level " + level + " using '" + statement + "'");
wasChanged = true;
} catch (SQLException exup) {
con.rollback();
String message = String.format(
"Attempted to upgrade %s to DB level %d using '%s', but failed with SQLException '%s' (code: %d/state: %s) - will continue without.",
tableName, level, statement, exup.getMessage(), exup.getErrorCode(), exup.getSQLState());
if (LOG.isDebugEnabled()) {
LOG.debug(message, exup);
} else {
LOG.info(message);
}
} finally {
closeStatement(upgradeStatement);
}
}
return wasChanged;
}
private boolean addModifiedIndex(Connection con, String tableName) throws SQLException {
boolean wasChanged = false;
String statement = this.dbInfo.getModifiedIndexStatement(tableName);
Statement upgradeStatement = null;
try {
upgradeStatement = con.createStatement();
upgradeStatement.execute(statement);
upgradeStatement.close();
con.commit();
LOG.info("Added 'modified' index to " + tableName + " using '" + statement + "'");
wasChanged = true;
} catch (SQLException exup) {
con.rollback();
String message = String.format(
"Attempted to add 'modified' index to %s using '%s', but failed with SQLException '%s' (code: %d/state: %s) - will continue without.",
tableName, statement, exup.getMessage(), exup.getErrorCode(), exup.getSQLState());
if (LOG.isDebugEnabled()) {
LOG.debug(message, exup);
} else {
LOG.info(message);
}
} finally {
closeStatement(upgradeStatement);
}
return wasChanged;
}
private static void getTableMetaData(Connection con, Collection<? extends Document> col, RDBTableMetaData tmd) throws SQLException {
Statement checkStatement = null;
ResultSet checkResultSet = null;
try {
checkStatement = con.createStatement();
checkResultSet = checkStatement.executeQuery("select * from " + tmd.getName() + " where ID = '0'");
// try to discover size of DATA column and binary-ness of ID
ResultSetMetaData met = checkResultSet.getMetaData();
obtainFlagsFromResultSetMeta(met, tmd);
String tableInfo = RDBJDBCTools.dumpResultSetMeta(met);
tmd.setSchemaInfo(tableInfo);
String indexInfo = dumpIndexData(con.getMetaData(), met, tmd.getName(), null);
tmd.setIndexInfo(indexInfo);
} finally {
closeResultSet(checkResultSet);
closeStatement(checkStatement);
}
}
public boolean isReadOnly() {
return readOnly;
}
@Override
protected void finalize() throws Throwable {
if (!this.ch.isClosed() && this.callStack != null) {
LOG.debug("finalizing RDBDocumentStore that was not disposed", this.callStack);
}
super.finalize();
}
private <T extends Document> T readDocumentCached(final Collection<T> collection, final String id, int maxCacheAge) {
if (collection != Collection.NODES) {
return readDocumentUncached(collection, id, null);
} else {
NodeDocument doc = null;
if (maxCacheAge > 0) {
// first try without lock
doc = nodesCache.getIfPresent(id);
if (doc != null) {
long lastCheckTime = doc.getLastCheckTime();
if (lastCheckTime != 0) {
if (maxCacheAge == Integer.MAX_VALUE || System.currentTimeMillis() - lastCheckTime < maxCacheAge) {
stats.doneFindCached(Collection.NODES, id);
return castAsT(unwrap(doc));
}
}
}
}
try {
try (CacheLock lock = acquireLockFor(id)) {
// caller really wants the cache to be cleared
if (maxCacheAge == 0) {
invalidateNodesCache(id, true);
doc = null;
}
final NodeDocument cachedDoc = doc;
doc = nodesCache.get(id, new Callable<NodeDocument>() {
@Override
public NodeDocument call() throws Exception {
NodeDocument doc = (NodeDocument) readDocumentUncached(collection, id, cachedDoc);
if (doc != null) {
doc.seal();
}
return wrap(doc);
}
});
// inspect the doc whether it can be used
long lastCheckTime = doc.getLastCheckTime();
if (lastCheckTime != 0 && (maxCacheAge == 0 || maxCacheAge == Integer.MAX_VALUE)) {
// we either just cleared the cache or the caller does
// not care;
} else if (lastCheckTime != 0 && (System.currentTimeMillis() - lastCheckTime < maxCacheAge)) {
// is new enough
} else {
// need to at least revalidate
NodeDocument ndoc = (NodeDocument) readDocumentUncached(collection, id, cachedDoc);
if (ndoc != null) {
ndoc.seal();
}
doc = wrap(ndoc);
nodesCache.put(doc);
}
}
return castAsT(unwrap(doc));
} catch (ExecutionException e) {
throw new IllegalStateException("Failed to load document with " + id, e);
}
}
}
@Nullable
private <T extends Document> boolean internalCreate(Collection<T> collection, List<UpdateOp> updates) {
final Stopwatch watch = startWatch();
List<String> ids = new ArrayList<String>(updates.size());
boolean success = true;
try {
// try up to CHUNKSIZE ops in one transaction
for (List<UpdateOp> chunks : Lists.partition(updates, CHUNKSIZE)) {
List<T> docs = new ArrayList<T>();
for (UpdateOp update : chunks) {
ids.add(update.getId());
maintainUpdateStats(collection, update.getId());
UpdateUtils.assertUnconditional(update);
T doc = collection.newDocument(this);
addUpdateCounters(update);
UpdateUtils.applyChanges(doc, update);
if (!update.getId().equals(doc.getId())) {
throw new DocumentStoreException("ID mismatch - UpdateOp: " + update.getId() + ", ID property: "
+ doc.getId());
}
docs.add(doc);
}
boolean done = insertDocuments(collection, docs);
if (done) {
if (collection == Collection.NODES) {
for (T doc : docs) {
nodesCache.putIfAbsent((NodeDocument) doc);
}
}
} else {
success = false;
}
}
return success;
} catch (DocumentStoreException ex) {
return false;
} finally {
stats.doneCreate(watch.elapsed(TimeUnit.NANOSECONDS), collection, ids, success);
}
}
@Nullable
private <T extends Document> T internalCreateOrUpdate(Collection<T> collection, UpdateOp update, boolean allowCreate,
boolean checkConditions, int retries) {
T oldDoc = readDocumentCached(collection, update.getId(), Integer.MAX_VALUE);
if (oldDoc == null) {
if (!allowCreate || !update.isNew()) {
return null;
}
T doc = collection.newDocument(this);
if (checkConditions && !checkConditions(doc, update.getConditions())) {
return null;
}
addUpdateCounters(update);
UpdateUtils.applyChanges(doc, update);
try {
Stopwatch watch = startWatch();
if (!insertDocuments(collection, Collections.singletonList(doc))) {
throw new DocumentStoreException("Can't insert the document: " + doc.getId());
}
if (collection == Collection.NODES) {
nodesCache.putIfAbsent((NodeDocument) doc);
}
stats.doneFindAndModify(watch.elapsed(TimeUnit.NANOSECONDS), collection, update.getId(), true, true, 0);
return oldDoc;
} catch (DocumentStoreException ex) {
// may have failed due to a race condition; try update instead
// this is an edge case, so it's ok to bypass the cache
// (avoiding a race condition where the DB is already updated
// but the cache is not)
oldDoc = readDocumentUncached(collection, update.getId(), null);
if (oldDoc == null) {
// something else went wrong
LOG.error("insert failed, but document " + update.getId() + " is not present, aborting", ex);
throw (ex);
}
return internalUpdate(collection, update, oldDoc, checkConditions, retries);
}
} else {
T result = internalUpdate(collection, update, oldDoc, checkConditions, retries);
if (allowCreate && result == null) {
if (retries > 0) {
result = internalCreateOrUpdate(collection, update, allowCreate, checkConditions, retries - 1);
}
else {
LOG.error("update of " + update.getId() + " failed, race condition?");
throw new DocumentStoreException("update of " + update.getId() + " failed, race condition?", null,
DocumentStoreException.Type.TRANSIENT);
}
}
return result;
}
}
/**
* @return previous version of document or <code>null</code>
*/
@Nullable
private <T extends Document> T internalUpdate(Collection<T> collection, UpdateOp update, T oldDoc, boolean checkConditions,
int maxRetries) {
if (checkConditions && !UpdateUtils.checkConditions(oldDoc, update.getConditions())) {
return null;
} else {
maintainUpdateStats(collection, update.getId());
addUpdateCounters(update);
T doc = createNewDocument(collection, oldDoc, update);
final Stopwatch watch = startWatch();
boolean success = false;
int retries = maxRetries;
try (CacheLock lock = acquireLockFor(update.getId())) {
while (!success && retries > 0) {
long lastmodcount = modcountOf(oldDoc);
success = updateDocument(collection, doc, update, lastmodcount);
if (!success) {
retries -= 1;
oldDoc = readDocumentCached(collection, update.getId(), Integer.MAX_VALUE);
if (oldDoc != null) {
long newmodcount = modcountOf(oldDoc);
if (lastmodcount == newmodcount) {
// cached copy did not change so it probably was
// updated by a different instance, get a fresh one
LOG.debug("suspect update from different instance (current modcount: {}), refetching: {}...",
newmodcount, update.getId());
if (collection == Collection.NODES) {
nodesCache.invalidate(update.getId());
}
oldDoc = readDocumentUncached(collection, update.getId(), null);
if (oldDoc == null) {
LOG.debug("after refetch: {} is gone", update.getId());
} else {
LOG.debug("after refetch: modcount for {} is {}", update.getId(), modcountOf(oldDoc));
}
}
}
if (oldDoc == null) {
// document was there but is now gone
LOG.debug("failed to apply update because document is gone in the meantime: " + update.getId(), new Exception("call stack"));
return null;
}
if (checkConditions && !UpdateUtils.checkConditions(oldDoc, update.getConditions())) {
return null;
}
else {
addUpdateCounters(update);
doc = createNewDocument(collection, oldDoc, update);
}
} else {
if (collection == Collection.NODES) {
nodesCache.replaceCachedDocument((NodeDocument) oldDoc, (NodeDocument) doc);
}
}
}
if (!success) {
throw new DocumentStoreException("failed update of " + doc.getId() + " (race?) after " + maxRetries
+ " retries", null, DocumentStoreException.Type.TRANSIENT);
}
return oldDoc;
} finally {
int numOfAttempts = maxRetries - retries - 1;
stats.doneFindAndModify(watch.elapsed(TimeUnit.NANOSECONDS), collection,
update.getId(), false, success, numOfAttempts);
}
}
}
@NotNull
private <T extends Document> T createNewDocument(Collection<T> collection, T oldDoc, UpdateOp update) {
T doc = collection.newDocument(this);
oldDoc.deepCopy(doc);
UpdateUtils.applyChanges(doc, update);
doc.seal();
return doc;
}
private static void addUpdateCounters(UpdateOp update) {
if (hasChangesToCollisions(update)) {
update.increment(COLLISIONSMODCOUNT, 1);
}
update.increment(MODCOUNT, 1);
}
private <T extends Document> List<T> internalQuery(Collection<T> collection, String fromKey, String toKey,
List<String> excludeKeyPatterns, List<QueryCondition> conditions, int limit) {
Connection connection = null;
RDBTableMetaData tmd = getTable(collection);
for (QueryCondition cond : conditions) {
if (!INDEXEDPROPERTIES.contains(cond.getPropertyName())) {
String message = "indexed property " + cond.getPropertyName() + " not supported, query was '" + cond
+ "'; supported properties are " + INDEXEDPROPERTIES;
LOG.info(message);
throw new DocumentStoreException(message);
}
}
final Stopwatch watch = startWatch();
int resultSize = 0;
try (CacheChangesTracker tracker = obtainTracker(collection, fromKey, toKey)) {
long now = System.currentTimeMillis();
connection = this.ch.getROConnection();
String from = collection == Collection.NODES && NodeDocument.MIN_ID_VALUE.equals(fromKey) ? null : fromKey;
String to = collection == Collection.NODES && NodeDocument.MAX_ID_VALUE.equals(toKey) ? null : toKey;
// OAK-6839: only populate the cache with *new* entries if the query
// isn't open-ended (something done by GC processes)
boolean populateCache = to != null;
List<RDBRow> dbresult = db.query(connection, tmd, from, to, excludeKeyPatterns, conditions, limit);
connection.commit();
int size = dbresult.size();
List<T> result = new ArrayList<T>(size);
for (int i = 0; i < size; i++) {
// free RDBRow as early as possible
RDBRow row = dbresult.set(i, null);
T doc = getIfCached(collection, row.getId(), row.getModcount());
if (doc == null) {
// parse DB contents into document if and only if it's not
// already in the cache
doc = convertFromDBObject(collection, row);
} else {
// we got a document from the cache, thus collection is NODES
// and a tracker is present
long lastmodified = modifiedOf(doc);
if (lastmodified == row.getModified() && lastmodified >= 1) {
try (CacheLock lock = acquireLockFor(row.getId())) {
if (!tracker.mightBeenAffected(row.getId())) {
// otherwise mark it as fresh
((NodeDocument) doc).markUpToDate(now);
}
}
}
else {
// we need a fresh document instance
doc = convertFromDBObject(collection, row);
}
}
result.add(doc);
}
if (collection == Collection.NODES) {
if (populateCache) {
nodesCache.putNonConflictingDocs(tracker, castAsNodeDocumentList(result));
} else {
Map<String, ModificationStamp> invMap = Maps.newHashMap();
for (Document doc : result) {
invMap.put(doc.getId(), new ModificationStamp(modcountOf(doc), modifiedOf(doc)));
}
nodesCache.invalidateOutdated(invMap);
}
}
resultSize = result.size();
return result;
} catch (Exception ex) {
LOG.error("SQL exception on query", ex);
throw asDocumentStoreException(ex, "SQL exception on query");
} finally {
this.ch.closeConnection(connection);
stats.doneQuery(watch.elapsed(TimeUnit.NANOSECONDS), collection, fromKey, toKey,
!conditions.isEmpty(), resultSize, -1, false);
}
}
private static interface MyCloseableIterable<T> extends Closeable, Iterable<T> {
}
protected <T extends Document> Iterable<T> queryAsIterable(final Collection<T> collection, String fromKey, String toKey,
final List<String> excludeKeyPatterns, final List<QueryCondition> conditions, final int limit, final String sortBy) {
final RDBTableMetaData tmd = getTable(collection);
Set<String> allowedProps = Sets.intersection(INDEXEDPROPERTIES, tmd.getColumnProperties());
for (QueryCondition cond : conditions) {
if (!allowedProps.contains(cond.getPropertyName())) {
String message = "indexed property " + cond.getPropertyName() + " not supported, query was '" + cond
+ "'; supported properties are " + allowedProps;
LOG.info(message);
throw new UnsupportedIndexedPropertyException(message);
}
}
final String from = collection == Collection.NODES && NodeDocument.MIN_ID_VALUE.equals(fromKey) ? null : fromKey;
final String to = collection == Collection.NODES && NodeDocument.MAX_ID_VALUE.equals(toKey) ? null : toKey;
return new MyCloseableIterable<T>() {
Set<Iterator<RDBRow>> returned = Sets.newHashSet();
@Override
public Iterator<T> iterator() {
try {
Iterator<RDBRow> res = db.queryAsIterator(ch, tmd, from, to, excludeKeyPatterns, conditions,
limit, sortBy);
returned.add(res);
Iterator<T> tmp = Iterators.transform(res, new Function<RDBRow, T>() {
@Override
public T apply(RDBRow input) {
return convertFromDBObject(collection, input);
}
});
return CloseableIterator.wrap(tmp, (Closeable) res);
} catch (SQLException ex) {
throw new RuntimeException(ex);
}
}
@Override
public void close() throws IOException {
for (Iterator<RDBRow> rdbi : returned) {
if (rdbi instanceof Closeable) {
((Closeable) rdbi).close();
}
}
}
};
}
protected <T extends Document> long queryCount(final Collection<T> collection, String fromKey, String toKey,
final List<String> excludeKeyPatterns, final List<QueryCondition> conditions) {
return internalGetAggregate(collection, "COUNT", "*", fromKey, toKey, excludeKeyPatterns, conditions);
}
protected <T extends Document> long getMinValue(final Collection<T> collection, String field, String fromKey, String toKey,
final List<String> excludeKeyPatterns, final List<QueryCondition> conditions) {
return internalGetAggregate(collection, "MIN", field, fromKey, toKey, excludeKeyPatterns, conditions);
}
private <T extends Document> long internalGetAggregate(final Collection<T> collection, final String aggregrate, String field,
String fromKey, String toKey, final List<String> excludeKeyPatterns, final List<QueryCondition> conditions) {
final RDBTableMetaData tmd = getTable(collection);
for (QueryCondition cond : conditions) {
if (!INDEXEDPROPERTIES.contains(cond.getPropertyName())) {
String message = "indexed property " + cond.getPropertyName() + " not supported, query was '" + cond
+ "'; supported properties are " + INDEXEDPROPERTIES;
LOG.info(message);
throw new DocumentStoreException(message);
}
}
final String from = collection == Collection.NODES && NodeDocument.MIN_ID_VALUE.equals(fromKey) ? null : fromKey;
final String to = collection == Collection.NODES && NodeDocument.MAX_ID_VALUE.equals(toKey) ? null : toKey;
Connection connection = null;
try {
connection = ch.getROConnection();
long result = db.getLong(connection, tmd, aggregrate, field, from, to, excludeKeyPatterns, conditions);
connection.commit();
return result;
} catch (SQLException ex) {
LOG.error("SQL exception on query", ex);
throw asDocumentStoreException(ex, "SQL exception on query");
} finally {
this.ch.closeConnection(connection);
}
}
@NotNull
protected <T extends Document> RDBTableMetaData getTable(Collection<T> collection) {
RDBTableMetaData tmd = this.tableMeta.get(collection);
if (tmd != null) {
return tmd;
} else {
throw new IllegalArgumentException("Unknown collection: " + collection.toString());
}
}
@Nullable
private <T extends Document> T readDocumentUncached(Collection<T> collection, String id, NodeDocument cachedDoc) {
Connection connection = null;
RDBTableMetaData tmd = getTable(collection);
final Stopwatch watch = startWatch();
boolean docFound = true;
try {
long lastmodcount = -1, lastmodified = -1;
if (cachedDoc != null) {
lastmodcount = modcountOf(cachedDoc);
lastmodified = modifiedOf(cachedDoc);
}
connection = this.ch.getROConnection();
RDBRow row = db.read(connection, tmd, id, lastmodcount, lastmodified);
connection.commit();
if (row == null) {
docFound = false;
return null;
} else {
if (lastmodcount == row.getModcount() && lastmodified == row.getModified() && lastmodified >= 1) {
// we can re-use the cached document
cachedDoc.markUpToDate(System.currentTimeMillis());
return castAsT(cachedDoc);
} else {
return convertFromDBObject(collection, row);
}
}
} catch (Exception ex) {
throw asDocumentStoreException(ex, "exception while reading " + id);
} finally {
this.ch.closeConnection(connection);
stats.doneFindUncached(watch.elapsed(TimeUnit.NANOSECONDS), collection, id, docFound, false);
}
}
private <T extends Document> void delete(Collection<T> collection, String id) {
Connection connection = null;
RDBTableMetaData tmd = getTable(collection);
Stopwatch watch = startWatch();
try {
connection = this.ch.getRWConnection();
db.delete(connection, tmd, Collections.singletonList(id));
connection.commit();
} catch (Exception ex) {
this.ch.rollbackConnection(connection);
throw handleException("removing " + id, ex, collection, id);
} finally {
this.ch.closeConnection(connection);
stats.doneRemove(watch.elapsed(TimeUnit.NANOSECONDS), collection, 1);
}
}
private <T extends Document> int delete(Collection<T> collection, List<String> ids) {
int numDeleted = 0;
RDBTableMetaData tmd = getTable(collection);
for (List<String> sublist : Lists.partition(ids, 64)) {
Connection connection = null;
Stopwatch watch = startWatch();
try {
connection = this.ch.getRWConnection();
numDeleted += db.delete(connection, tmd, sublist);
connection.commit();
} catch (Exception ex) {
this.ch.rollbackConnection(connection);
throw handleException("removing " + ids, ex, collection, ids);
} finally {
this.ch.closeConnection(connection);
stats.doneRemove(watch.elapsed(TimeUnit.NANOSECONDS), collection, ids.size());
}
}
return numDeleted;
}
private <T extends Document> int delete(Collection<T> collection, Map<String, Long> toRemove) {
int numDeleted = 0;
RDBTableMetaData tmd = getTable(collection);
Map<String, Long> subMap = Maps.newHashMap();
Iterator<Entry<String, Long>> it = toRemove.entrySet().iterator();
while (it.hasNext()) {
Entry<String, Long> entry = it.next();
subMap.put(entry.getKey(), entry.getValue());
if (subMap.size() == 64 || !it.hasNext()) {
Connection connection = null;
int num = 0;
Stopwatch watch = startWatch();
try {
connection = this.ch.getRWConnection();
num = db.delete(connection, tmd, subMap);
numDeleted += num;
connection.commit();
} catch (Exception ex) {
this.ch.rollbackConnection(connection);
Set<String> ids = subMap.keySet();
throw handleException("deleting " + ids, ex, collection, ids);
} finally {
this.ch.closeConnection(connection);
stats.doneRemove(watch.elapsed(TimeUnit.NANOSECONDS), collection, num);
}
subMap.clear();
}
}
return numDeleted;
}
private <T extends Document> int deleteWithCondition(Collection<T> collection, List<QueryCondition> conditions) {
int numDeleted = 0;
RDBTableMetaData tmd = getTable(collection);
Stopwatch watch = startWatch();
Connection connection = null;
try {
connection = this.ch.getRWConnection();
numDeleted = db.deleteWithCondition(connection, tmd, conditions);
connection.commit();
} catch (Exception ex) {
this.ch.rollbackConnection(connection);
throw asDocumentStoreException(ex, "deleting " + collection + ": " + conditions);
} finally {
this.ch.closeConnection(connection);
stats.doneRemove(watch.elapsed(TimeUnit.NANOSECONDS), collection, numDeleted);
}
return numDeleted;
}
private <T extends Document> boolean updateDocument(@NotNull Collection<T> collection, @NotNull T document,
@NotNull UpdateOp update, Long oldmodcount) {
Connection connection = null;
RDBTableMetaData tmd = getTable(collection);
String data = null;
try {
connection = this.ch.getRWConnection();
Number hasBinary = (Number) document.get(NodeDocument.HAS_BINARY_FLAG);
Boolean deletedOnce = (Boolean) document.get(NodeDocument.DELETED_ONCE);
Long modcount = (Long) document.get(MODCOUNT);
Long cmodcount = (Long) document.get(COLLISIONSMODCOUNT);
boolean success = false;
boolean shouldRetry = true;
// every 16th update is a full rewrite
if (isAppendableUpdate(update) && modcount % 16 != 0) {
String appendData = ser.asString(update, tmd.getColumnOnlyProperties());
if (appendData.length() < tmd.getDataLimitInOctets() / CHAR2OCTETRATIO) {
try {
Operation modOperation = update.getChanges().get(MODIFIEDKEY);
long modified = getModifiedFromOperation(modOperation);
boolean modifiedIsConditional = modOperation == null || modOperation.type != UpdateOp.Operation.Type.SET;
success = db.appendingUpdate(connection, tmd, document.getId(), modified, modifiedIsConditional, hasBinary,
deletedOnce, modcount, cmodcount, oldmodcount, appendData);
// if we get here, a retry is not going to help (the SQL
// operation succeeded but simply did not select a row
// that could be updated, likely because of the check on
// MODCOUNT
shouldRetry = false;
connection.commit();
} catch (SQLException ex) {
continueIfStringOverflow(ex);
this.ch.rollbackConnection(connection);
success = false;
}
}
}
if (!success && shouldRetry) {
data = ser.asString(document, tmd.getColumnOnlyProperties());
Object m = document.get(MODIFIED);
long modified = (m instanceof Long) ? ((Long)m).longValue() : 0;
success = db.update(connection, tmd, document.getId(), modified, hasBinary, deletedOnce, modcount, cmodcount,
oldmodcount, data);
connection.commit();
}
return success;
} catch (SQLException ex) {
this.ch.rollbackConnection(connection);
String addDiags = "";
if (data != null && RDBJDBCTools.matchesSQLState(ex, "22", "72")) {
byte[] bytes = asBytes(data);
addDiags = String.format(" (DATA size in Java characters: %d, in octets: %d, computed character limit: %d)",
data.length(), bytes.length, tmd.getDataLimitInOctets() / CHAR2OCTETRATIO);
}
String message = String.format("Update for %s failed%s", document.getId(), addDiags);
LOG.debug(message, ex);
throw handleException(message, ex, collection, document.getId());
} finally {
this.ch.closeConnection(connection);
}
}
private static void continueIfStringOverflow(SQLException ex) throws SQLException {
String state = ex.getSQLState();
if ("22001".equals(state) /* everybody */|| ("72000".equals(state) && 1489 == ex.getErrorCode()) /* Oracle */
|| ("S0001".equals(state) && 2628 == ex.getErrorCode()) /* MSSQL update*/) {
// ok
} else {
throw (ex);
}
}
private static boolean isAppendableUpdate(UpdateOp update) {
return NOAPPEND == false;
}
private static long getModifiedFromOperation(Operation op) {
return op == null ? 0L : Long.parseLong(op.value.toString());
}
private <T extends Document> boolean insertDocuments(Collection<T> collection, List<T> documents) {
Connection connection = null;
RDBTableMetaData tmd = getTable(collection);
try {
connection = this.ch.getRWConnection();
Set<String> insertedKeys = db.insert(connection, tmd, documents);
connection.commit();
return insertedKeys.size() == documents.size();
} catch (SQLException ex) {
this.ch.rollbackConnection(connection);
List<String> ids = new ArrayList<String>();
for (T doc : documents) {
ids.add(doc.getId());
}
String message = String.format("insert of %s failed", ids);
LOG.debug(message, ex);
// collect additional exceptions
String messages = LOG.isDebugEnabled() ? RDBJDBCTools.getAdditionalMessages(ex) : "";
// see whether a DATA error was involved
boolean dataRelated = false;
SQLException walk = ex;
while (walk != null && !dataRelated) {
dataRelated = RDBJDBCTools.matchesSQLState(walk, "22", "72");
walk = walk.getNextException();
}
if (dataRelated) {
String id = null;
int longest = 0, longestChars = 0;
for (Document d : documents) {
String data = ser.asString(d, tmd.getColumnOnlyProperties());
byte bytes[] = asBytes(data);
if (bytes.length > longest) {
longest = bytes.length;
longestChars = data.length();
id = d.getId();
}
}
String m = String
.format(" (potential cause: long data for ID %s - longest octet DATA size in Java characters: %d, in octets: %d, computed character limit: %d)",
id, longest, longestChars, tmd.getDataLimitInOctets() / CHAR2OCTETRATIO);
messages += m;
}
if (!messages.isEmpty()) {
LOG.debug("additional diagnostics: " + messages);
}
throw handleException(message, ex, collection, ids);
} finally {
this.ch.closeConnection(connection);
}
}
// configuration
// Whether to use GZIP compression
private static final boolean NOGZIP = SystemPropertySupplier
.create("org.apache.jackrabbit.oak.plugins.document.rdb.RDBDocumentStore.NOGZIP", Boolean.FALSE).loggingTo(LOG).get();
// Whether to use append operations (string concatenation) in the DATA column
private static final boolean NOAPPEND = SystemPropertySupplier
.create("org.apache.jackrabbit.oak.plugins.document.rdb.RDBDocumentStore.NOAPPEND", Boolean.FALSE).loggingTo(LOG).get();
// Number of documents to insert at once for batch create
private static final int CHUNKSIZE = SystemPropertySupplier
.create("org.apache.jackrabbit.oak.plugins.document.rdb.RDBDocumentStore.CHUNKSIZE", 64).loggingTo(LOG)
.validateWith(value -> value > 0).get();
// Number of query hits above which a diagnostic warning is generated
private static final int QUERYHITSLIMIT = SystemPropertySupplier
.create("org.apache.jackrabbit.oak.plugins.document.rdb.RDBDocumentStore.QUERYHITSLIMIT", 4096).loggingTo(LOG)
.validateWith(value -> value > 0).get();
// Number of elapsed ms in a query above which a diagnostic warning is generated
private static final int QUERYTIMELIMIT = SystemPropertySupplier
.create("org.apache.jackrabbit.oak.plugins.document.rdb.RDBDocumentStore.QUERYTIMELIMIT", 10000).loggingTo(LOG)
.validateWith(value -> value > 0).get();
// Whether to use JDBC batch commands for the createOrUpdate (default: true)
private static final boolean BATCHUPDATES = SystemPropertySupplier
.create(RDBDocumentStore.class.getName() + ".BATCHUPDATES", Boolean.TRUE).loggingTo(LOG)
.formatSetMessage((name, value) -> {
return String.format("Batch updates disabled (system property %s set to '%s')", name, value);
}).get();
public static byte[] asBytes(@NotNull String data) {
byte[] bytes;
try {
bytes = data.getBytes("UTF-8");
} catch (UnsupportedEncodingException ex) {
LOG.error("UTF-8 not supported??", ex);
throw asDocumentStoreException(ex, "UTF-8 not supported??");
}
if (NOGZIP) {
return bytes;
} else {
ByteArrayOutputStream bos = new ByteArrayOutputStream(bytes.length / 2);
try (GZIPOutputStream gos = asGZIPOutputStream(bos, Deflater.BEST_SPEED)) {
gos.write(bytes);
gos.close();
byte[] compressedBytes = bos.toByteArray();
if (LOG.isTraceEnabled()) {
long ratio = (100L * compressedBytes.length) / bytes.length;
LOG.trace("Gzipped {} bytes to {} ({}%)", bytes.length, compressedBytes.length, ratio);
}
return compressedBytes;
} catch (IOException ex) {
LOG.error("Error while gzipping contents", ex);
throw asDocumentStoreException(ex, "Error while gzipping contents");
}
}
}
private static GZIPOutputStream asGZIPOutputStream(OutputStream os, final int level) throws IOException {
return new GZIPOutputStream(os) {
{
this.def.setLevel(level);
}
};
}
@Override
public void setReadWriteMode(String readWriteMode) {
// ignored
}
public void setStatsCollector(DocumentStoreStatsCollector stats) {
this.stats = stats;
}
@SuppressWarnings("unchecked")
private static <T extends Document> T castAsT(NodeDocument doc) {
return (T) doc;
}
@SuppressWarnings("unchecked")
private static <T extends Document> List<NodeDocument> castAsNodeDocumentList(List<T> list) {
return (List<NodeDocument>) list;
}
private NodeDocumentCache nodesCache;
private NodeDocumentLocks locks;
@Nullable
private static NodeDocument unwrap(@NotNull NodeDocument doc) {
return doc == NodeDocument.NULL ? null : doc;
}
@NotNull
private static NodeDocument wrap(@Nullable NodeDocument doc) {
return doc == null ? NodeDocument.NULL : doc;
}
@NotNull
private static String idOf(@NotNull Document doc) {
String id = doc.getId();
if (id == null) {
throw new IllegalArgumentException("non-null ID expected");
}
return id;
}
private static long modcountOf(@NotNull Document doc) {
Long n = doc.getModCount();
return n != null ? n : -1;
}
private static long modifiedOf(@NotNull Document doc) {
Object l = doc.get(NodeDocument.MODIFIED_IN_SECS);
return (l instanceof Long) ? ((Long)l).longValue() : -1;
}
@NotNull
protected <T extends Document> T convertFromDBObject(@NotNull Collection<T> collection, @NotNull RDBRow row) {
// this method is present here in order to facilitate unit testing for OAK-3566
return ser.fromRow(collection, row);
}
private static boolean hasChangesToCollisions(UpdateOp update) {
if (!USECMODCOUNT) {
return false;
} else {
for (Entry<Key, Operation> e : checkNotNull(update).getChanges().entrySet()) {
Key k = e.getKey();
Operation op = e.getValue();
if (op.type == Operation.Type.SET_MAP_ENTRY) {
if (NodeDocument.COLLISIONS.equals(k.getName())) {
return true;
}
}
}
return false;
}
}
@NotNull
private static <T extends Document> String dumpKeysAndModcounts(Map<String, T> docs) {
if (docs.isEmpty()) {
return "-";
} else {
StringBuilder sb = new StringBuilder();
for (Map.Entry<String, T> e : docs.entrySet()) {
Long mc = e.getValue().getModCount();
if (sb.length() != 0) {
sb.append(", ");
}
sb.append(String.format("%s (%s)", e.getKey(), mc == null ? "" : mc.toString()));
}
return sb.toString();
}
}
// keeping track of CLUSTER_NODES updates
private Map<String, Long> cnUpdates = new ConcurrentHashMap<String, Long>();
private <T extends Document> void maintainUpdateStats(Collection<T> collection, String key) {
if (collection == Collection.CLUSTER_NODES) {
synchronized (this) {
Long old = cnUpdates.get(key);
old = old == null ? Long.valueOf(1) : old + 1;
cnUpdates.put(key, old);
}
}
}
private String getCnStats() {
if (cnUpdates.isEmpty()) {
return "";
} else {
List<Map.Entry<String, Long>> tmp = new ArrayList<>(cnUpdates.entrySet());
Collections.sort(tmp, (Entry<String, Long> o1, Entry<String, Long> o2) -> o1.getKey().compareTo(o2.getKey()));
return " (Cluster Node updates: " + tmp.toString() + ")";
}
}
private Stopwatch startWatch() {
return Stopwatch.createStarted();
}
protected NodeDocumentCache getNodeDocumentCache() {
return nodesCache;
}
private <T extends Document> DocumentStoreException handleException(String message, Exception ex, Collection<T> collection,
java.util.Collection<String> ids) {
if (collection == Collection.NODES) {
for (String id : ids) {
invalidateCache(collection, id, false);
}
}
return asDocumentStoreException(ex, message);
}
private <T extends Document> DocumentStoreException handleException(String message, Exception ex, Collection<T> collection,
String id) {
return handleException(message, ex, collection, Collections.singleton(id));
}
protected class UnsupportedIndexedPropertyException extends DocumentStoreException {
private static final long serialVersionUID = -8392572622365260105L;
public UnsupportedIndexedPropertyException(String message) {
super(message);
}
}
private CacheLock acquireLockFor(String id) {
return new CacheLock(this.locks, id);
}
private static class CacheLock implements AutoCloseable {
private final Lock lock;
public CacheLock(NodeDocumentLocks locks, String id) {
this.lock = locks.acquire(id);
}
@Override
public void close() {
lock.unlock();
}
}
// slightly extended query support
protected static class QueryCondition {
private final String propertyName, operator;
private final List<? extends Object> operands;
public QueryCondition(String propertyName, String operator, long value) {
this.propertyName = propertyName;
this.operator = operator;
this.operands = Collections.singletonList(value);
}
public QueryCondition(String propertyName, String operator, List<? extends Object> values) {
this.propertyName = propertyName;
this.operator = operator;
this.operands = values;
}
public QueryCondition(String propertyName, String operator) {
this.propertyName = propertyName;
this.operator = operator;
this.operands = Collections.emptyList();
}
public String getPropertyName() {
return propertyName;
}
public String getOperator() {
return operator;
}
public List<? extends Object> getOperands() {
return this.operands;
}
@Override
public String toString() {
if (this.operands.isEmpty()) {
return String.format("%s %s", propertyName, operator);
} else if (this.operands.size() == 1) {
return String.format("%s %s %s", propertyName, operator, operands.get(0).toString());
} else {
return String.format("%s %s %s", propertyName, operator, operands.toString());
}
}
}
}