You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by st...@apache.org on 2009/06/28 07:47:37 UTC
svn commit: r789051 [2/4] - in /hadoop/hbase/trunk: conf/ src/contrib/
src/contrib/transactional/ src/contrib/transactional/src/
src/contrib/transactional/src/java/ src/contrib/transactional/src/java/org/
src/contrib/transactional/src/java/org/apache/ ...
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionScannerCallable.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionScannerCallable.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionScannerCallable.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionScannerCallable.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,50 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client.transactional;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.client.HConnection;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.ScannerCallable;
+import org.apache.hadoop.hbase.ipc.TransactionalRegionInterface;
+
+class TransactionScannerCallable extends ScannerCallable {
+
+ private TransactionState transactionState;
+
+ TransactionScannerCallable(final TransactionState transactionState,
+ final HConnection connection, final byte[] tableName,
+ final byte[] startRow, Scan scan) {
+ super(connection, tableName, startRow, scan);
+ this.transactionState = transactionState;
+ }
+
+ @Override
+ protected long openScanner() throws IOException {
+ if (transactionState.addRegion(location)) {
+ ((TransactionalRegionInterface) server).beginTransaction(transactionState
+ .getTransactionId(), location.getRegionInfo().getRegionName());
+ }
+ return ((TransactionalRegionInterface) server).openScanner(transactionState
+ .getTransactionId(), this.location.getRegionInfo().getRegionName(),
+ getScan());
+ }
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,78 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client.transactional;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.HRegionLocation;
+
+/**
+ * Holds client-side transaction information. Client's use them as opaque
+ * objects passed around to transaction operations.
+ *
+ */
+public class TransactionState {
+ static final Log LOG = LogFactory.getLog(TransactionState.class);
+
+ private final long transactionId;
+
+ private Set<HRegionLocation> participatingRegions = new HashSet<HRegionLocation>();
+
+ TransactionState(final long transactionId) {
+ this.transactionId = transactionId;
+ }
+
+ boolean addRegion(final HRegionLocation hregion) {
+ boolean added = participatingRegions.add(hregion);
+
+ if (added) {
+ LOG.debug("Adding new hregion ["
+ + hregion.getRegionInfo().getRegionNameAsString()
+ + "] to transaction [" + transactionId + "]");
+ }
+
+ return added;
+ }
+
+ Set<HRegionLocation> getParticipatingRegions() {
+ return participatingRegions;
+ }
+
+ /**
+ * Get the transactionId.
+ *
+ * @return Return the transactionId.
+ */
+ public long getTransactionId() {
+ return transactionId;
+ }
+
+ /**
+ * @see java.lang.Object#toString()
+ */
+ @Override
+ public String toString() {
+ return "id: " + transactionId + ", particpants: "
+ + participatingRegions.size();
+ }
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,190 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client.transactional;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HConnection;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.ScannerCallable;
+import org.apache.hadoop.hbase.client.ServerCallable;
+import org.apache.hadoop.hbase.ipc.TransactionalRegionInterface;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Table with transactional support.
+ *
+ */
+public class TransactionalTable extends HTable {
+
+ /**
+ * @param conf
+ * @param tableName
+ * @throws IOException
+ */
+ public TransactionalTable(final HBaseConfiguration conf,
+ final String tableName) throws IOException {
+ this(conf, Bytes.toBytes(tableName));
+ }
+
+ /**
+ * @param conf
+ * @param tableName
+ * @throws IOException
+ */
+ public TransactionalTable(final HBaseConfiguration conf,
+ final byte[] tableName) throws IOException {
+ super(conf, tableName);
+ }
+
+ private static abstract class TransactionalServerCallable<T> extends
+ ServerCallable<T> {
+ protected TransactionState transactionState;
+
+ protected TransactionalRegionInterface getTransactionServer() {
+ return (TransactionalRegionInterface) server;
+ }
+
+ protected void recordServer() throws IOException {
+ if (transactionState.addRegion(location)) {
+ getTransactionServer().beginTransaction(
+ transactionState.getTransactionId(),
+ location.getRegionInfo().getRegionName());
+ }
+ }
+
+ /**
+ * @param connection
+ * @param tableName
+ * @param row
+ * @param transactionState
+ */
+ public TransactionalServerCallable(final HConnection connection,
+ final byte[] tableName, final byte[] row,
+ final TransactionState transactionState) {
+ super(connection, tableName, row);
+ this.transactionState = transactionState;
+ }
+
+ }
+
+ /**
+ * Method for getting data from a row
+ * @param get the Get to fetch
+ * @return the result
+ * @throws IOException
+ * @since 0.20.0
+ */
+ public Result get(final TransactionState transactionState, final Get get) throws IOException {
+ return super.getConnection().getRegionServerWithRetries(
+ new TransactionalServerCallable<Result>(super.getConnection(), super
+ .getTableName(), get.getRow(), transactionState) {
+ public Result call() throws IOException {
+ recordServer();
+ return getTransactionServer().get(
+ transactionState.getTransactionId(),
+ location.getRegionInfo().getRegionName(), get);
+ }
+ });
+ }
+
+
+ /**
+ *
+ * @param delete
+ * @throws IOException
+ * @since 0.20.0
+ */
+ public void delete(final TransactionState transactionState, final Delete delete)
+ throws IOException {
+ super.getConnection().getRegionServerWithRetries(
+ new TransactionalServerCallable<Object>(super.getConnection(), super
+ .getTableName(), delete.getRow(), transactionState) {
+ public Object call() throws IOException {
+ recordServer();
+ getTransactionServer().delete(
+ transactionState.getTransactionId(),
+ location.getRegionInfo().getRegionName(), delete);
+ return null;
+ }
+ });
+
+ }
+
+ /**
+ * Commit a Put to the table.
+ * <p>
+ * If autoFlush is false, the update is buffered.
+ * @param put
+ * @throws IOException
+ * @since 0.20.0
+ */
+ public synchronized void put(TransactionState transactionState, final Put put) throws IOException {
+ //super.validatePut(put);
+ super.getConnection().getRegionServerWithRetries(
+ new TransactionalServerCallable<Object>(super.getConnection(), super
+ .getTableName(), put.getRow(), transactionState) {
+ public Object call() throws IOException {
+ recordServer();
+ getTransactionServer().put(
+ transactionState.getTransactionId(),
+ location.getRegionInfo().getRegionName(), put);
+ return null;
+ }
+ });
+
+ }
+
+ public ResultScanner getScanner(final TransactionState transactionState,
+ Scan scan) throws IOException {
+ ClientScanner scanner = new TransactionalClientScanner(transactionState, scan);
+ scanner.initialize();
+ return scanner;
+ }
+
+ protected class TransactionalClientScanner extends HTable.ClientScanner {
+
+ private TransactionState transactionState;
+
+ protected TransactionalClientScanner(
+ final TransactionState transactionState, Scan scan) {
+ super(scan);
+ this.transactionState = transactionState;
+ }
+
+ @Override
+ protected ScannerCallable getScannerCallable(
+ final byte[] localStartKey, int caching) {
+ TransactionScannerCallable t =
+ new TransactionScannerCallable(transactionState, getConnection(),
+ getTableName(), getScan().getStartRow(), getScan());
+ t.setCaching(caching);
+ return t;
+ }
+ }
+
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/UnknownTransactionException.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/UnknownTransactionException.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/UnknownTransactionException.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/UnknownTransactionException.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,43 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.client.transactional;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+
+/**
+ * Thrown if a region server is passed an unknown transaction id
+ */
+public class UnknownTransactionException extends DoNotRetryIOException {
+
+ private static final long serialVersionUID = 698575374929591099L;
+
+ /** constructor */
+ public UnknownTransactionException() {
+ super();
+ }
+
+ /**
+ * Constructor
+ * @param s message
+ */
+ public UnknownTransactionException(String s) {
+ super(s);
+ }
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/package.html
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/package.html?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/package.html (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/client/transactional/package.html Sun Jun 28 05:47:35 2009
@@ -0,0 +1,61 @@
+<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">
+<html>
+
+<!--
+ 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.
+-->
+
+<head />
+<body bgcolor="white">
+
+This package provides support for atomic transactions. Transactions can
+span multiple regions. Transaction writes are applied when committing a
+transaction. At commit time, the transaction is examined to see if it
+can be applied while still maintaining atomicity. This is done by
+looking for conflicts with the transactions that committed while the
+current transaction was running. This technique is known as optimistic
+concurrency control (OCC) because it relies on the assumption that
+transactions will mostly not have conflicts with each other.
+
+<p>
+For more details on OCC, see the paper <i> On Optimistic Methods for Concurrency Control </i>
+by Kung and Robinson available
+<a href=http://www.seas.upenn.edu/~zives/cis650/papers/opt-cc.pdf> here </a>.
+
+<p> To enable transactions, modify hbase-site.xml to turn on the
+TransactionalRegionServer. This is done by setting
+<i>hbase.regionserver.class</i> to
+<i>org.apache.hadoop.hbase.ipc.TransactionalRegionInterface</i> and
+<i>hbase.regionserver.impl </i> to
+<i>org.apache.hadoop.hbase.regionserver.transactional.TransactionalRegionServer</i>
+
+<p>
+The read set claimed by a transactional scanner is determined from the start and
+ end keys which the scanner is opened with.
+
+
+
+<h3> Known Issues </h3>
+
+Recovery in the face of hregion server failure
+is not fully implemented. Thus, you cannot rely on the transactional
+properties in the face of node failure.
+
+
+
+
+</body>
+</html>
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/ipc/IndexedRegionInterface.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/ipc/IndexedRegionInterface.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/ipc/IndexedRegionInterface.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/ipc/IndexedRegionInterface.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,11 @@
+/*
+ * $Id$
+ * Created on Sep 10, 2008
+ *
+ */
+package org.apache.hadoop.hbase.ipc;
+
+/** Interface for the indexed region server. */
+public interface IndexedRegionInterface extends TransactionalRegionInterface {
+ // No methods for now...
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/ipc/TransactionalRegionInterface.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/ipc/TransactionalRegionInterface.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/ipc/TransactionalRegionInterface.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/ipc/TransactionalRegionInterface.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.ipc;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+
+/**
+ * Interface for transactional region servers.
+ *
+ * <p>
+ * NOTE: if you change the interface, you must change the RPC version number in
+ * HBaseRPCProtocolVersion
+ *
+ */
+public interface TransactionalRegionInterface extends HRegionInterface {
+
+ /** Status code representing a transaction that can be committed. */
+ int COMMIT_OK = 1;
+ /** Status code representing a read-only transaction that can be committed. */
+ int COMMIT_OK_READ_ONLY = 2;
+ /** Status code representing a transaction that cannot be committed. */
+ int COMMIT_UNSUCESSFUL = 3;
+
+ /**
+ * Sent to initiate a transaction.
+ *
+ * @param transactionId
+ * @param regionName name of region
+ * @throws IOException
+ */
+ void beginTransaction(long transactionId, final byte[] regionName)
+ throws IOException;
+
+ /**
+ * Perform a transactional Get operation.
+ * @param regionName name of region to get from
+ * @param get Get operation
+ * @return Result
+ * @throws IOException
+ */
+ public Result get(long transactionId, byte [] regionName, Get get) throws IOException;
+
+
+ /**
+ * Transactional put data into the specified region
+ * @param regionName
+ * @param put the data to be put
+ * @throws IOException
+ */
+ public void put(long transactionId, final byte [] regionName, final Put put)
+ throws IOException;
+
+ /**
+ * Put an array of puts into the specified region
+ * @param regionName
+ * @param puts
+ * @return
+ * @throws IOException
+ */
+ public int put(long transactionId, final byte[] regionName, final Put [] puts)
+ throws IOException;
+
+
+ /**
+ * Deletes all the KeyValues that match those found in the Delete object,
+ * if their ts <= to the Delete. In case of a delete with a specific ts it
+ * only deletes that specific KeyValue.
+ * @param regionName
+ * @param delete
+ * @throws IOException
+ */
+ public void delete(long transactionId, final byte[] regionName, final Delete delete)
+ throws IOException;
+
+ //
+ // remote scanner interface
+ //
+
+ /**
+ * Opens a remote transactional scanner with a RowFilter.
+ *
+ * @param regionName name of region to scan
+ * @param scan configured scan object
+ * @return scannerId scanner identifier used in other calls
+ * @throws IOException
+ */
+ public long openScanner(long transactionId, final byte [] regionName, final Scan scan)
+ throws IOException;
+
+ /**
+ * Ask if we can commit the given transaction.
+ *
+ * @param regionName
+ * @param transactionId
+ * @return status of COMMIT_OK, COMMIT_READ_ONLY, or COMMIT_UNSUSESSFULL
+ * @throws IOException
+ */
+ int commitRequest(final byte[] regionName, long transactionId)
+ throws IOException;
+
+ /**
+ * Try to commit the given transaction. This is used when there is only one
+ * participating region.
+ *
+ * @param regionName
+ * @param transactionId
+ * @return true if committed
+ * @throws IOException
+ */
+ boolean commitIfPossible(final byte[] regionName, long transactionId)
+ throws IOException;
+
+ /**
+ * Commit the transaction.
+ *
+ * @param regionName
+ * @param transactionId
+ * @throws IOException
+ */
+ void commit(final byte[] regionName, long transactionId) throws IOException;
+
+ /**
+ * Abort the transaction.
+ *
+ * @param regionName
+ * @param transactionId
+ * @throws IOException
+ */
+ void abort(final byte[] regionName, long transactionId) throws IOException;
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexMaintenanceUtils.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexMaintenanceUtils.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexMaintenanceUtils.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexMaintenanceUtils.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,82 @@
+/**
+ * Copyright 2008 The Apache Software Foundation Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific language governing permissions and limitations under the
+ * License.
+ */
+package org.apache.hadoop.hbase.regionserver.tableindexed;
+
+import java.util.SortedMap;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.ColumnNameParseException;
+import org.apache.hadoop.hbase.HStoreKey;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.tableindexed.IndexSpecification;
+import org.apache.hadoop.hbase.client.tableindexed.IndexedTable;
+import org.apache.hadoop.hbase.io.BatchUpdate;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Singleton class for index maintence logic.
+ */
+public class IndexMaintenanceUtils {
+
+ private static final Log LOG = LogFactory.getLog(IndexMaintenanceUtils.class);
+
+ public static Put createIndexUpdate(final IndexSpecification indexSpec, final byte[] row,
+ final SortedMap<byte[], byte[]> columnValues) {
+ byte[] indexRow = indexSpec.getKeyGenerator().createIndexKey(row, columnValues);
+ Put update = new Put(indexRow);
+
+ update.add(IndexedTable.INDEX_COL_FAMILY_NAME, IndexedTable.INDEX_BASE_ROW, row);
+
+ try {
+ for (byte[] col : indexSpec.getIndexedColumns()) {
+ byte[] val = columnValues.get(col);
+ if (val == null) {
+ throw new RuntimeException("Unexpected missing column value. [" + Bytes.toString(col) + "]");
+ }
+ byte [][] colSeperated = HStoreKey.parseColumn(col);
+ update.add(colSeperated[0], colSeperated[1], val);
+ }
+
+ for (byte[] col : indexSpec.getAdditionalColumns()) {
+ byte[] val = columnValues.get(col);
+ if (val != null) {
+ byte [][] colSeperated = HStoreKey.parseColumn(col);
+ update.add(colSeperated[0], colSeperated[1], val);
+ }
+ }
+ } catch (ColumnNameParseException e) {
+ throw new RuntimeException(e);
+ }
+
+ return update;
+ }
+
+ /**
+ * Ask if this update does apply to the index.
+ *
+ * @param indexSpec
+ * @param b
+ * @return true if possibly apply.
+ */
+ public static boolean doesApplyToIndex(final IndexSpecification indexSpec,
+ final SortedMap<byte[], byte[]> columnValues) {
+
+ for (byte[] neededCol : indexSpec.getIndexedColumns()) {
+ if (!columnValues.containsKey(neededCol)) {
+ LOG.debug("Index [" + indexSpec.getIndexId() + "] can't be updated because ["
+ + Bytes.toString(neededCol) + "] is missing");
+ return false;
+ }
+ }
+ return true;
+ }
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegion.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegion.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegion.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegion.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,289 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.regionserver.tableindexed;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.NavigableSet;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.Map.Entry;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.Leases;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.tableindexed.IndexSpecification;
+import org.apache.hadoop.hbase.client.tableindexed.IndexedTableDescriptor;
+import org.apache.hadoop.hbase.io.BatchUpdate;
+import org.apache.hadoop.hbase.regionserver.FlushRequester;
+import org.apache.hadoop.hbase.regionserver.HLog;
+import org.apache.hadoop.hbase.regionserver.transactional.TransactionalRegion;
+import org.apache.hadoop.hbase.util.Bytes;
+
+class IndexedRegion extends TransactionalRegion {
+
+ private static final Log LOG = LogFactory.getLog(IndexedRegion.class);
+
+ private final HBaseConfiguration conf;
+ private final IndexedTableDescriptor indexTableDescriptor;
+ private Map<IndexSpecification, HTable> indexSpecToTable = new HashMap<IndexSpecification, HTable>();
+
+ public IndexedRegion(final Path basedir, final HLog log, final FileSystem fs,
+ final HBaseConfiguration conf, final HRegionInfo regionInfo,
+ final FlushRequester flushListener, Leases trxLeases) throws IOException {
+ super(basedir, log, fs, conf, regionInfo, flushListener, trxLeases);
+ this.indexTableDescriptor = new IndexedTableDescriptor(regionInfo.getTableDesc());
+ this.conf = conf;
+ }
+
+ private synchronized HTable getIndexTable(IndexSpecification index)
+ throws IOException {
+ HTable indexTable = indexSpecToTable.get(index);
+ if (indexTable == null) {
+ indexTable = new HTable(conf, index.getIndexedTableName(super
+ .getRegionInfo().getTableDesc().getName()));
+ indexSpecToTable.put(index, indexTable);
+ }
+ return indexTable;
+ }
+
+ private Collection<IndexSpecification> getIndexes() {
+ return indexTableDescriptor.getIndexes();
+ }
+
+ /**
+ * @param batchUpdate
+ * @param lockid
+ * @param writeToWAL if true, then we write this update to the log
+ * @throws IOException
+ */
+ @Override
+ public void put(Put put, Integer lockId, boolean writeToWAL)
+ throws IOException {
+ updateIndexes(put); // Do this first because will want to see the old row
+ super.put(put, lockId, writeToWAL);
+ }
+
+ private void updateIndexes(Put put) throws IOException {
+ List<IndexSpecification> indexesToUpdate = new LinkedList<IndexSpecification>();
+
+ // Find the indexes we need to update
+ for (IndexSpecification index : getIndexes()) {
+ if (possiblyAppliesToIndex(index, put)) {
+ indexesToUpdate.add(index);
+ }
+ }
+
+ if (indexesToUpdate.size() == 0) {
+ return;
+ }
+
+ NavigableSet<byte[]> neededColumns = getColumnsForIndexes(indexesToUpdate);
+ NavigableMap<byte[], byte[]> newColumnValues = getColumnsFromPut(put);
+
+ Get oldGet = new Get(put.getRow());
+ for (byte [] neededCol : neededColumns) {
+ oldGet.addColumn(neededCol);
+ }
+
+ Result oldResult = super.get(oldGet, null);
+
+ // Add the old values to the new if they are not there
+ if (oldResult != null && oldResult.raw() != null) {
+ for (KeyValue oldKV : oldResult.raw()) {
+ if (!newColumnValues.containsKey(oldKV.getColumn())) {
+ newColumnValues.put(oldKV.getColumn(), oldKV.getValue());
+ }
+ }
+ }
+
+ Iterator<IndexSpecification> indexIterator = indexesToUpdate.iterator();
+ while (indexIterator.hasNext()) {
+ IndexSpecification indexSpec = indexIterator.next();
+ if (!IndexMaintenanceUtils.doesApplyToIndex(indexSpec, newColumnValues)) {
+ indexIterator.remove();
+ }
+ }
+
+ SortedMap<byte[], byte[]> oldColumnValues = convertToValueMap(oldResult);
+
+ for (IndexSpecification indexSpec : indexesToUpdate) {
+ removeOldIndexEntry(indexSpec, put.getRow(), oldColumnValues);
+ updateIndex(indexSpec, put.getRow(), newColumnValues);
+ }
+ }
+
+ private void updateIndexes(Delete delete) {
+ // FIXME
+ // Handle delete batch updates. Go back and get the next older values
+// for (BatchOperation op : batchUpdate) {
+// if (!op.isPut()) {
+// Cell current = oldColumnCells.get(op.getColumn());
+// if (current != null) {
+// // TODO: Fix this profligacy!!! St.Ack
+// Cell [] older = Cell.createSingleCellArray(super.get(batchUpdate.getRow(),
+// op.getColumn(), current.getTimestamp(), 1));
+// if (older != null && older.length > 0) {
+// newColumnValues.put(op.getColumn(), older[0].getValue());
+// }
+// }
+// }
+// }
+ }
+
+ /** Return the columns needed for the update. */
+ private NavigableSet<byte[]> getColumnsForIndexes(Collection<IndexSpecification> indexes) {
+ NavigableSet<byte[]> neededColumns = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
+ for (IndexSpecification indexSpec : indexes) {
+ for (byte[] col : indexSpec.getAllColumns()) {
+ neededColumns.add(col);
+ }
+ }
+ return neededColumns;
+ }
+
+ private void removeOldIndexEntry(IndexSpecification indexSpec, byte[] row,
+ SortedMap<byte[], byte[]> oldColumnValues) throws IOException {
+ for (byte[] indexedCol : indexSpec.getIndexedColumns()) {
+ if (!oldColumnValues.containsKey(indexedCol)) {
+ LOG.debug("Index [" + indexSpec.getIndexId()
+ + "] not trying to remove old entry for row ["
+ + Bytes.toString(row) + "] because col ["
+ + Bytes.toString(indexedCol) + "] is missing");
+ return;
+ }
+ }
+
+ byte[] oldIndexRow = indexSpec.getKeyGenerator().createIndexKey(row,
+ oldColumnValues);
+ LOG.debug("Index [" + indexSpec.getIndexId() + "] removing old entry ["
+ + Bytes.toString(oldIndexRow) + "]");
+ getIndexTable(indexSpec).delete(new Delete(oldIndexRow));
+ }
+
+ private NavigableMap<byte[], byte[]> getColumnsFromPut(Put put) {
+ NavigableMap<byte[], byte[]> columnValues = new TreeMap<byte[], byte[]>(
+ Bytes.BYTES_COMPARATOR);
+ for (List<KeyValue> familyPuts : put.getFamilyMap().values()) {
+ for (KeyValue kv : familyPuts) {
+ columnValues.put(kv.getColumn(), kv.getValue());
+ }
+ }
+ return columnValues;
+ }
+
+ /** Ask if this put *could* apply to the index. It may actually apply if some of the columns needed are missing.
+ *
+ * @param indexSpec
+ * @param put
+ * @return true if possibly apply.
+ */
+ private boolean possiblyAppliesToIndex(IndexSpecification indexSpec, Put put) {
+ for (List<KeyValue> familyPuts : put.getFamilyMap().values()) {
+ for (KeyValue kv : familyPuts) {
+ if (indexSpec.containsColumn(kv.getColumn())) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ // FIXME: This call takes place in an RPC, and requires an RPC. This makes for
+ // a likely deadlock if the number of RPCs we are trying to serve is >= the
+ // number of handler threads.
+ private void updateIndex(IndexSpecification indexSpec, byte[] row,
+ SortedMap<byte[], byte[]> columnValues) throws IOException {
+ Put indexUpdate = IndexMaintenanceUtils.createIndexUpdate(indexSpec, row, columnValues);
+ getIndexTable(indexSpec).put(indexUpdate);
+ LOG.debug("Index [" + indexSpec.getIndexId() + "] adding new entry ["
+ + Bytes.toString(indexUpdate.getRow()) + "] for row ["
+ + Bytes.toString(row) + "]");
+
+ }
+
+ @Override
+ public void delete(Delete delete, final Integer lockid, boolean writeToWAL)
+ throws IOException {
+
+ if (!getIndexes().isEmpty()) {
+ // Need all columns
+ NavigableSet<byte[]> neededColumns = getColumnsForIndexes(getIndexes());
+
+ Get get = new Get(delete.getRow());
+ for (byte [] col : neededColumns) {
+ get.addColumn(col);
+ }
+
+ Result oldRow = super.get(get, null);
+ SortedMap<byte[], byte[]> oldColumnValues = convertToValueMap(oldRow);
+
+
+ for (IndexSpecification indexSpec : getIndexes()) {
+ removeOldIndexEntry(indexSpec, delete.getRow(), oldColumnValues);
+ }
+
+ // Handle if there is still a version visible.
+ if (delete.getTimeStamp() != HConstants.LATEST_TIMESTAMP) {
+ get.setTimeRange(1, delete.getTimeStamp());
+ oldRow = super.get(get, null);
+ SortedMap<byte[], byte[]> currentColumnValues = convertToValueMap(oldRow);
+
+ for (IndexSpecification indexSpec : getIndexes()) {
+ if (IndexMaintenanceUtils.doesApplyToIndex(indexSpec, currentColumnValues)) {
+ updateIndex(indexSpec, delete.getRow(), currentColumnValues);
+ }
+ }
+ }
+ }
+ super.delete(delete, lockid, writeToWAL);
+ }
+
+ private SortedMap<byte[], byte[]> convertToValueMap(Result result) {
+ SortedMap<byte[], byte[]> currentColumnValues = new TreeMap<byte[], byte[]>(Bytes.BYTES_COMPARATOR);
+
+ if (result == null || result.raw() == null) {
+ return currentColumnValues;
+ }
+
+ for(KeyValue kv : result.list()) {
+ currentColumnValues.put(kv.getColumn(), kv.getValue());
+ }
+ return currentColumnValues;
+ }
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegionServer.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegionServer.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegionServer.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/tableindexed/IndexedRegionServer.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,67 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.regionserver.tableindexed;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.ipc.HBaseRPCProtocolVersion;
+import org.apache.hadoop.hbase.ipc.IndexedRegionInterface;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.transactional.TransactionalRegionServer;
+import org.apache.hadoop.util.Progressable;
+
+/**
+ * RegionServer which maintains secondary indexes.
+ *
+ **/
+public class IndexedRegionServer extends TransactionalRegionServer implements
+ IndexedRegionInterface {
+
+ public IndexedRegionServer(HBaseConfiguration conf) throws IOException {
+ super(conf);
+ }
+
+ @Override
+ public long getProtocolVersion(final String protocol, final long clientVersion)
+ throws IOException {
+ if (protocol.equals(IndexedRegionInterface.class.getName())) {
+ return HBaseRPCProtocolVersion.versionID;
+ }
+ return super.getProtocolVersion(protocol, clientVersion);
+ }
+
+ @Override
+ protected HRegion instantiateRegion(final HRegionInfo regionInfo)
+ throws IOException {
+ HRegion r = new IndexedRegion(HTableDescriptor.getTableDir(super
+ .getRootDir(), regionInfo.getTableDesc().getName()), super.hlog, super
+ .getFileSystem(), super.conf, regionInfo, super.getFlushRequester(), super.getTransactionalLeases());
+ r.initialize(null, new Progressable() {
+ public void progress() {
+ addProcessingMessage(regionInfo);
+ }
+ });
+ return r;
+ }
+
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/CleanOldTransactionsChore.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/CleanOldTransactionsChore.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/CleanOldTransactionsChore.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/CleanOldTransactionsChore.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,57 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.regionserver.transactional;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.hadoop.hbase.Chore;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+
+/**
+ * Cleans up committed transactions when they are no longer needed to verify
+ * pending transactions.
+ */
+class CleanOldTransactionsChore extends Chore {
+
+ private static final String SLEEP_CONF = "hbase.transaction.clean.sleep";
+ private static final int DEFAULT_SLEEP = 60 * 1000;
+
+ private final TransactionalRegionServer regionServer;
+
+ /**
+ * @param regionServer
+ * @param stopRequest
+ */
+ public CleanOldTransactionsChore(
+ final TransactionalRegionServer regionServer,
+ final AtomicBoolean stopRequest) {
+ super(regionServer.getConfiguration().getInt(SLEEP_CONF, DEFAULT_SLEEP),
+ stopRequest);
+ this.regionServer = regionServer;
+ }
+
+ @Override
+ protected void chore() {
+ for (HRegion region : regionServer.getOnlineRegions()) {
+ ((TransactionalRegion) region).removeUnNeededCommitedTransactions();
+ }
+ }
+
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLog.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLog.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLog.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLog.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,117 @@
+package org.apache.hadoop.hbase.regionserver.transactional;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.regionserver.HLog;
+import org.apache.hadoop.hbase.regionserver.HLogKey;
+import org.apache.hadoop.hbase.regionserver.LogRollListener;
+import org.apache.hadoop.io.SequenceFile;
+
+/**
+ * Add support for transactional operations to the regionserver's
+ * write-ahead-log.
+ *
+ */
+class THLog extends HLog {
+
+ public THLog(FileSystem fs, Path dir, Configuration conf,
+ LogRollListener listener) throws IOException {
+ super(fs, dir, conf, listener);
+ }
+
+ @Override
+ protected SequenceFile.Writer createWriter(Path path) throws IOException {
+ return super.createWriter(path, THLogKey.class, KeyValue.class);
+ }
+
+ @Override
+ protected HLogKey makeKey(byte[] regionName, byte[] tableName, long seqNum,
+ long now) {
+ return new THLogKey(regionName, tableName, seqNum, now);
+ }
+
+ public void writeStartToLog(HRegionInfo regionInfo, final long transactionId) throws IOException {
+ this.append(regionInfo, System.currentTimeMillis(),
+ THLogKey.TrxOp.START, transactionId);
+ }
+
+ public void writeUpdateToLog(HRegionInfo regionInfo, final long transactionId, final Put update)
+ throws IOException {
+ this.append(regionInfo, update, transactionId);
+ }
+
+ public void writeDeleteToLog(HRegionInfo regionInfo, final long transactionId, final Delete delete)
+ throws IOException {
+ // FIXME
+ }
+
+ public void writeCommitToLog(HRegionInfo regionInfo, final long transactionId) throws IOException {
+ this.append(regionInfo, System.currentTimeMillis(),
+ THLogKey.TrxOp.COMMIT, transactionId);
+ }
+
+ public void writeAbortToLog(HRegionInfo regionInfo, final long transactionId) throws IOException {
+ this.append(regionInfo, System.currentTimeMillis(),
+ THLogKey.TrxOp.ABORT, transactionId);
+ }
+
+ /**
+ * Write a general transaction op to the log. This covers: start, commit, and
+ * abort.
+ *
+ * @param regionInfo
+ * @param now
+ * @param txOp
+ * @param transactionId
+ * @throws IOException
+ */
+ public void append(HRegionInfo regionInfo, long now, THLogKey.TrxOp txOp,
+ long transactionId) throws IOException {
+ THLogKey key = new THLogKey(regionInfo.getRegionName(), regionInfo
+ .getTableDesc().getName(), -1, now, txOp, transactionId);
+ super.append(regionInfo, key, new KeyValue(new byte [0], 0, 0)); // Empty KeyValue
+ }
+
+ /**
+ * Write a transactional update to the log.
+ *
+ * @param regionInfo
+ * @param now
+ * @param update
+ * @param transactionId
+ * @throws IOException
+ */
+ public void append(HRegionInfo regionInfo, Put update, long transactionId)
+ throws IOException {
+
+ long commitTime = System.currentTimeMillis(); // FIXME ?
+
+ THLogKey key = new THLogKey(regionInfo.getRegionName(), regionInfo
+ .getTableDesc().getName(), -1, commitTime, THLogKey.TrxOp.OP,
+ transactionId);
+
+ for (KeyValue value : convertToKeyValues(update)) {
+ super.append(regionInfo, key, value);
+ }
+ }
+
+ private List<KeyValue> convertToKeyValues(Put update) {
+ List<KeyValue> edits = new ArrayList<KeyValue>();
+
+ for (List<KeyValue> kvs : update.getFamilyMap().values()) {
+ for (KeyValue kv : kvs) {
+ edits.add(kv);
+ }
+ }
+ return edits;
+ }
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLogKey.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLogKey.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLogKey.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLogKey.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,79 @@
+package org.apache.hadoop.hbase.regionserver.transactional;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.regionserver.HLogKey;
+
+public class THLogKey extends HLogKey {
+
+ /** Type of Transactional op going into the HLot
+ *
+ */
+ public enum TrxOp {
+ /** Start a transaction. */
+ START((byte)1),
+ /** A standard operation that is transactional. KV holds the op. */
+ OP((byte)2),
+ /** A transaction was committed. */
+ COMMIT((byte)3),
+ /** A transaction was aborted. */
+ ABORT((byte)4);
+
+ private final byte opCode;
+
+ private TrxOp(byte opCode) {
+ this.opCode = opCode;
+ }
+
+ public static TrxOp fromByte(byte opCode) {
+ for (TrxOp op : TrxOp.values()) {
+ if (op.opCode == opCode) {
+ return op;
+ }
+ }
+ return null;
+ }
+
+ }
+
+ private byte transactionOp = -1;
+ private long transactionId = -1;
+
+ public THLogKey() {
+ // For Writable
+ }
+
+ public THLogKey(byte[] regionName, byte[] tablename, long logSeqNum, long now) {
+ super(regionName, tablename, logSeqNum, now);
+ }
+
+ public THLogKey(byte[] regionName, byte[] tablename, long logSeqNum, long now, TrxOp op, long transactionId) {
+ super(regionName, tablename, logSeqNum, now);
+ this.transactionOp = op.opCode;
+ this.transactionId = transactionId;
+ }
+
+ public TrxOp getTrxOp() {
+ return TrxOp.fromByte(this.transactionOp);
+ }
+
+ public long getTransactionId() {
+ return this.transactionId;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ super.write(out);
+ out.writeByte(transactionOp);
+ out.writeLong(transactionId);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ super.readFields(in);
+ this.transactionOp = in.readByte();
+ this.transactionId = in.readLong();
+ }
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLogRecoveryManager.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLogRecoveryManager.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLogRecoveryManager.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/THLogRecoveryManager.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,254 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.regionserver.transactional;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.Map.Entry;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.transactional.TransactionLogger;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.util.Progressable;
+
+/**
+ * Responsible recovering transactional information from the HLog.
+ */
+class THLogRecoveryManager {
+ private static final Log LOG = LogFactory
+ .getLog(THLogRecoveryManager.class);
+
+ private final FileSystem fileSystem;
+ private final HRegionInfo regionInfo;
+ private final HBaseConfiguration conf;
+
+ /**
+ * @param region
+ */
+ public THLogRecoveryManager(final TransactionalRegion region) {
+ this.fileSystem = region.getFilesystem();
+ this.regionInfo = region.getRegionInfo();
+ this.conf = region.getConf();
+ }
+
+ // For Testing
+ THLogRecoveryManager(final FileSystem fileSystem,
+ final HRegionInfo regionInfo, final HBaseConfiguration conf) {
+ this.fileSystem = fileSystem;
+ this.regionInfo = regionInfo;
+ this.conf = conf;
+ }
+
+
+
+ /**
+ * @param reconstructionLog
+ * @param maxSeqID
+ * @param reporter
+ * @return map of batch updates
+ * @throws UnsupportedEncodingException
+ * @throws IOException
+ */
+ public Map<Long, List<KeyValue>> getCommitsFromLog(
+ final Path reconstructionLog, final long maxSeqID,
+ final Progressable reporter) throws UnsupportedEncodingException,
+ IOException {
+ if (reconstructionLog == null || !fileSystem.exists(reconstructionLog)) {
+ // Nothing to do.
+ return null;
+ }
+ // Check its not empty.
+ FileStatus[] stats = fileSystem.listStatus(reconstructionLog);
+ if (stats == null || stats.length == 0) {
+ LOG.warn("Passed reconstruction log " + reconstructionLog
+ + " is zero-length");
+ return null;
+ }
+
+ SortedMap<Long, List<KeyValue>> pendingTransactionsById = new TreeMap<Long, List<KeyValue>>();
+ SortedMap<Long, List<KeyValue>> commitedTransactionsById = new TreeMap<Long, List<KeyValue>>();
+ Set<Long> abortedTransactions = new HashSet<Long>();
+
+ SequenceFile.Reader logReader = new SequenceFile.Reader(fileSystem,
+ reconstructionLog, conf);
+
+ try {
+ THLogKey key = new THLogKey();
+ KeyValue val = new KeyValue();
+ long skippedEdits = 0;
+ long totalEdits = 0;
+ long startCount = 0;
+ long writeCount = 0;
+ long abortCount = 0;
+ long commitCount = 0;
+ // How many edits to apply before we send a progress report.
+ int reportInterval = conf.getInt("hbase.hstore.report.interval.edits",
+ 2000);
+
+ while (logReader.next(key, val)) {
+ LOG.debug("Processing edit: key: " + key.toString() + " val: "
+ + val.toString());
+ if (key.getLogSeqNum() < maxSeqID) {
+ skippedEdits++;
+ continue;
+ }
+
+ if (key.getTrxOp() == null || !Bytes.equals(key.getRegionName(), regionInfo.getRegionName())) {
+ continue;
+ }
+ long transactionId = key.getTransactionId();
+
+ List<KeyValue> updates = pendingTransactionsById.get(transactionId);
+ switch (key.getTrxOp()) {
+
+ case START:
+ if (updates != null || abortedTransactions.contains(transactionId)
+ || commitedTransactionsById.containsKey(transactionId)) {
+ LOG.error("Processing start for transaction: " + transactionId
+ + ", but have already seen start message");
+ throw new IOException("Corrupted transaction log");
+ }
+ updates = new LinkedList<KeyValue>();
+ pendingTransactionsById.put(transactionId, updates);
+ startCount++;
+ break;
+
+ case OP:
+ if (updates == null) {
+ LOG.error("Processing edit for transaction: " + transactionId
+ + ", but have not seen start message");
+ throw new IOException("Corrupted transaction log");
+ }
+
+ updates.add(val);
+ val = new KeyValue();
+ writeCount++;
+ break;
+
+ case ABORT:
+ if (updates == null) {
+ LOG.error("Processing abort for transaction: " + transactionId
+ + ", but have not seen start message");
+ throw new IOException("Corrupted transaction log");
+ }
+ abortedTransactions.add(transactionId);
+ pendingTransactionsById.remove(transactionId);
+ abortCount++;
+ break;
+
+ case COMMIT:
+ if (updates == null) {
+ LOG.error("Processing commit for transaction: " + transactionId
+ + ", but have not seen start message");
+ throw new IOException("Corrupted transaction log");
+ }
+ if (abortedTransactions.contains(transactionId)) {
+ LOG.error("Processing commit for transaction: " + transactionId
+ + ", but also have abort message");
+ throw new IOException("Corrupted transaction log");
+ }
+ if (updates.size() == 0) {
+ LOG
+ .warn("Transaciton " + transactionId
+ + " has no writes in log. ");
+ }
+ if (commitedTransactionsById.containsKey(transactionId)) {
+ LOG.error("Processing commit for transaction: " + transactionId
+ + ", but have already commited transaction with that id");
+ throw new IOException("Corrupted transaction log");
+ }
+ pendingTransactionsById.remove(transactionId);
+ commitedTransactionsById.put(transactionId, updates);
+ commitCount++;
+ break;
+ default:
+ throw new IllegalStateException("Unexpected log entry type");
+ }
+ totalEdits++;
+
+ if (reporter != null && (totalEdits % reportInterval) == 0) {
+ reporter.progress();
+ }
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Read " + totalEdits + " tranasctional operations (skipped "
+ + skippedEdits + " because sequence id <= " + maxSeqID + "): "
+ + startCount + " starts, " + writeCount + " writes, " + abortCount
+ + " aborts, and " + commitCount + " commits.");
+ }
+ } finally {
+ logReader.close();
+ }
+
+ if (pendingTransactionsById.size() > 0) {
+ resolvePendingTransaction(pendingTransactionsById, commitedTransactionsById);
+ }
+
+
+ return commitedTransactionsById;
+ }
+
+ private void resolvePendingTransaction(
+ SortedMap<Long, List<KeyValue>> pendingTransactionsById,
+ SortedMap<Long, List<KeyValue>> commitedTransactionsById) {
+ LOG.info("Region log has " + pendingTransactionsById.size()
+ + " unfinished transactions. Going to the transaction log to resolve");
+
+ for (Entry<Long, List<KeyValue>> entry : pendingTransactionsById
+ .entrySet()) {
+ TransactionLogger.TransactionStatus transactionStatus = getGlobalTransactionLog()
+ .getStatusForTransaction(entry.getKey());
+ if (transactionStatus == null) {
+ throw new RuntimeException("Cannot resolve tranasction ["
+ + entry.getKey() + "] from global tx log.");
+ }
+ switch (transactionStatus) {
+ case ABORTED:
+ break;
+ case COMMITTED:
+ commitedTransactionsById.put(entry.getKey(), entry.getValue());
+ break;
+ case PENDING:
+ break;
+ }
+ }
+ }
+
+ private TransactionLogger getGlobalTransactionLog() {
+ return null;
+ }
+
+
+}
Added: hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java
URL: http://svn.apache.org/viewvc/hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java?rev=789051&view=auto
==============================================================================
--- hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java (added)
+++ hadoop/hbase/trunk/src/contrib/transactional/src/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java Sun Jun 28 05:47:35 2009
@@ -0,0 +1,342 @@
+/**
+ * Copyright 2008 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.regionserver.transactional;
+
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.NavigableSet;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.Map.Entry;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Holds the state of a transaction.
+ */
+class TransactionState {
+
+ private static final Log LOG = LogFactory.getLog(TransactionState.class);
+
+ /** Current status. */
+ public enum Status {
+ /** Initial status, still performing operations. */
+ PENDING,
+ /**
+ * Checked if we can commit, and said yes. Still need to determine the
+ * global decision.
+ */
+ COMMIT_PENDING,
+ /** Committed. */
+ COMMITED,
+ /** Aborted. */
+ ABORTED
+ }
+
+ /**
+ * Simple container of the range of the scanners we've opened. Used to check
+ * for conflicting writes.
+ */
+ private static class ScanRange {
+ protected byte[] startRow;
+ protected byte[] endRow;
+
+ public ScanRange(byte[] startRow, byte[] endRow) {
+ this.startRow = startRow;
+ this.endRow = endRow;
+ }
+
+ /**
+ * Check if this scan range contains the given key.
+ *
+ * @param rowKey
+ * @return boolean
+ */
+ public boolean contains(byte[] rowKey) {
+ if (startRow != null && Bytes.compareTo(rowKey, startRow) < 0) {
+ return false;
+ }
+ if (endRow != null && Bytes.compareTo(endRow, rowKey) < 0) {
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "startRow: "
+ + (startRow == null ? "null" : Bytes.toString(startRow))
+ + ", endRow: " + (endRow == null ? "null" : Bytes.toString(endRow));
+ }
+ }
+
+ private final HRegionInfo regionInfo;
+ private final long hLogStartSequenceId;
+ private final long transactionId;
+ private Status status;
+ private SortedSet<byte[]> readSet = new TreeSet<byte[]>(
+ Bytes.BYTES_COMPARATOR);
+ private List<Put> writeSet = new LinkedList<Put>();
+ private List<ScanRange> scanSet = new LinkedList<ScanRange>();
+ private Set<TransactionState> transactionsToCheck = new HashSet<TransactionState>();
+ private int startSequenceNumber;
+ private Integer sequenceNumber;
+ private int commitPendingWaits = 0;
+
+ TransactionState(final long transactionId, final long rLogStartSequenceId,
+ HRegionInfo regionInfo) {
+ this.transactionId = transactionId;
+ this.hLogStartSequenceId = rLogStartSequenceId;
+ this.regionInfo = regionInfo;
+ this.status = Status.PENDING;
+ }
+
+ void addRead(final byte[] rowKey) {
+ readSet.add(rowKey);
+ }
+
+ Set<byte[]> getReadSet() {
+ return readSet;
+ }
+
+ void addWrite(final Put write) {
+ writeSet.add(write);
+ }
+
+ List<Put> getWriteSet() {
+ return writeSet;
+ }
+
+ void addDelete(final Delete delete) {
+ throw new UnsupportedOperationException("NYI");
+ }
+
+ /**
+ * GetFull from the writeSet.
+ *
+ * @param row
+ * @param columns
+ * @param timestamp
+ * @return
+ */
+ Result localGet(Get get) {
+
+ List<KeyValue> localKVs = new LinkedList<KeyValue>();
+
+ for (Put put : writeSet) {
+ if (!Bytes.equals(get.getRow(), put.getRow())) {
+ continue;
+ }
+ if (put.getTimeStamp() > get.getTimeRange().getMax()) {
+ continue;
+ }
+ if (put.getTimeStamp() < get.getTimeRange().getMin()) {
+ continue;
+ }
+
+ for (Entry<byte [], NavigableSet<byte []>> getFamilyEntry : get.getFamilyMap().entrySet()) {
+ List<KeyValue> familyPuts = put.getFamilyMap().get(getFamilyEntry.getKey());
+ if (familyPuts == null) {
+ continue;
+ }
+ if (getFamilyEntry.getValue() == null){
+ localKVs.addAll(familyPuts);
+ } else {
+ for (KeyValue kv : familyPuts) {
+ if (getFamilyEntry.getValue().contains(kv.getQualifier())) {
+ localKVs.add(kv);
+ }
+ }
+ }
+ }
+ }
+
+ if (localKVs.isEmpty()) {
+ return null;
+ }
+ return new Result(localKVs);
+ }
+
+ void addTransactionToCheck(final TransactionState transaction) {
+ transactionsToCheck.add(transaction);
+ }
+
+ boolean hasConflict() {
+ for (TransactionState transactionState : transactionsToCheck) {
+ if (hasConflict(transactionState)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private boolean hasConflict(final TransactionState checkAgainst) {
+ if (checkAgainst.getStatus().equals(TransactionState.Status.ABORTED)) {
+ return false; // Cannot conflict with aborted transactions
+ }
+
+ for (Put otherUpdate : checkAgainst.getWriteSet()) {
+ if (this.getReadSet().contains(otherUpdate.getRow())) {
+ LOG.debug("Transaction [" + this.toString()
+ + "] has read which conflicts with [" + checkAgainst.toString()
+ + "]: region [" + regionInfo.getRegionNameAsString() + "], row["
+ + Bytes.toString(otherUpdate.getRow()) + "]");
+ return true;
+ }
+ for (ScanRange scanRange : this.scanSet) {
+ if (scanRange.contains(otherUpdate.getRow())) {
+ LOG.debug("Transaction [" + this.toString()
+ + "] has scan which conflicts with [" + checkAgainst.toString()
+ + "]: region [" + regionInfo.getRegionNameAsString() + "], scanRange[" +
+ scanRange.toString()+"] ,row["
+ + Bytes.toString(otherUpdate.getRow()) + "]");
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Get the status.
+ *
+ * @return Return the status.
+ */
+ Status getStatus() {
+ return status;
+ }
+
+ /**
+ * Set the status.
+ *
+ * @param status The status to set.
+ */
+ void setStatus(final Status status) {
+ this.status = status;
+ }
+
+ /**
+ * Get the startSequenceNumber.
+ *
+ * @return Return the startSequenceNumber.
+ */
+ int getStartSequenceNumber() {
+ return startSequenceNumber;
+ }
+
+ /**
+ * Set the startSequenceNumber.
+ *
+ * @param startSequenceNumber
+ */
+ void setStartSequenceNumber(final int startSequenceNumber) {
+ this.startSequenceNumber = startSequenceNumber;
+ }
+
+ /**
+ * Get the sequenceNumber.
+ *
+ * @return Return the sequenceNumber.
+ */
+ Integer getSequenceNumber() {
+ return sequenceNumber;
+ }
+
+ /**
+ * Set the sequenceNumber.
+ *
+ * @param sequenceNumber The sequenceNumber to set.
+ */
+ void setSequenceNumber(final Integer sequenceNumber) {
+ this.sequenceNumber = sequenceNumber;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder result = new StringBuilder();
+ result.append("[transactionId: ");
+ result.append(transactionId);
+ result.append(" status: ");
+ result.append(status.name());
+ result.append(" read Size: ");
+ result.append(readSet.size());
+ result.append(" scan Size: ");
+ result.append(scanSet.size());
+ result.append(" write Size: ");
+ result.append(writeSet.size());
+ result.append(" startSQ: ");
+ result.append(startSequenceNumber);
+ if (sequenceNumber != null) {
+ result.append(" commitedSQ:");
+ result.append(sequenceNumber);
+ }
+ result.append("]");
+
+ return result.toString();
+ }
+
+ /**
+ * Get the transactionId.
+ *
+ * @return Return the transactionId.
+ */
+ long getTransactionId() {
+ return transactionId;
+ }
+
+ /**
+ * Get the startSequenceId.
+ *
+ * @return Return the startSequenceId.
+ */
+ long getHLogStartSequenceId() {
+ return hLogStartSequenceId;
+ }
+
+ void addScan(Scan scan) {
+ ScanRange scanRange = new ScanRange(scan.getStartRow(), scan.getStopRow());
+ LOG.trace(String.format(
+ "Adding scan for transcaction [%s], from [%s] to [%s]", transactionId,
+ scanRange.startRow == null ? "null" : Bytes
+ .toString(scanRange.startRow), scanRange.endRow == null ? "null"
+ : Bytes.toString(scanRange.endRow)));
+ scanSet.add(scanRange);
+ }
+
+ int getCommitPendingWaits() {
+ return commitPendingWaits;
+ }
+
+ void incrementCommitPendingWaits() {
+ this.commitPendingWaits++;
+ }
+
+}