| /* |
| * 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.phoenix.execute; |
| |
| import static com.google.common.base.Preconditions.checkNotNull; |
| import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT; |
| import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_SIZE; |
| import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BYTES; |
| import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_COMMIT_TIME; |
| import static org.apache.phoenix.query.QueryServices.WILDCARD_QUERY_DYNAMIC_COLS_ATTRIB; |
| import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_WILDCARD_QUERY_DYNAMIC_COLS_ATTRIB; |
| |
| import java.io.IOException; |
| import java.sql.SQLException; |
| import java.sql.Timestamp; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| |
| import javax.annotation.Nonnull; |
| import javax.annotation.concurrent.Immutable; |
| |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.TableName; |
| import org.apache.hadoop.hbase.client.Delete; |
| import org.apache.hadoop.hbase.client.Mutation; |
| import org.apache.hadoop.hbase.client.Put; |
| import org.apache.hadoop.hbase.client.Table; |
| import org.apache.hadoop.hbase.io.ImmutableBytesWritable; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.hbase.util.Pair; |
| import org.apache.htrace.Span; |
| import org.apache.htrace.TraceScope; |
| import org.apache.phoenix.cache.ServerCacheClient.ServerCache; |
| import org.apache.phoenix.compile.MutationPlan; |
| import org.apache.phoenix.coprocessor.BaseScannerRegionObserver; |
| import org.apache.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult; |
| import org.apache.phoenix.exception.SQLExceptionCode; |
| import org.apache.phoenix.exception.SQLExceptionInfo; |
| import org.apache.phoenix.hbase.index.IndexRegionObserver; |
| import org.apache.phoenix.hbase.index.exception.IndexWriteException; |
| import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr; |
| import org.apache.phoenix.index.IndexMaintainer; |
| import org.apache.phoenix.index.IndexMetaDataCacheClient; |
| import org.apache.phoenix.index.PhoenixIndexBuilder; |
| import org.apache.phoenix.index.PhoenixIndexFailurePolicy; |
| import org.apache.phoenix.index.PhoenixIndexFailurePolicy.MutateCommand; |
| import org.apache.phoenix.index.PhoenixIndexMetaData; |
| import org.apache.phoenix.jdbc.PhoenixConnection; |
| import org.apache.phoenix.jdbc.PhoenixStatement.Operation; |
| import org.apache.phoenix.monitoring.GlobalClientMetrics; |
| import org.apache.phoenix.monitoring.MutationMetricQueue; |
| import org.apache.phoenix.monitoring.MutationMetricQueue.MutationMetric; |
| import org.apache.phoenix.monitoring.MutationMetricQueue.NoOpMutationMetricsQueue; |
| import org.apache.phoenix.monitoring.ReadMetricQueue; |
| import org.apache.phoenix.query.QueryConstants; |
| import org.apache.phoenix.query.QueryServices; |
| import org.apache.phoenix.query.QueryServicesOptions; |
| import org.apache.phoenix.schema.IllegalDataException; |
| import org.apache.phoenix.schema.MetaDataClient; |
| import org.apache.phoenix.schema.PColumn; |
| import org.apache.phoenix.schema.PIndexState; |
| import org.apache.phoenix.schema.PMetaData; |
| import org.apache.phoenix.schema.PName; |
| import org.apache.phoenix.schema.PRow; |
| import org.apache.phoenix.schema.PTable; |
| import org.apache.phoenix.schema.PTableRef; |
| import org.apache.phoenix.schema.PTableType; |
| import org.apache.phoenix.schema.RowKeySchema; |
| import org.apache.phoenix.schema.TableNotFoundException; |
| import org.apache.phoenix.schema.TableRef; |
| import org.apache.phoenix.schema.ValueSchema.Field; |
| import org.apache.phoenix.schema.types.PLong; |
| import org.apache.phoenix.schema.types.PTimestamp; |
| import org.apache.phoenix.trace.util.Tracing; |
| import org.apache.phoenix.transaction.PhoenixTransactionContext; |
| import org.apache.phoenix.transaction.PhoenixTransactionContext.PhoenixVisibilityLevel; |
| import org.apache.phoenix.transaction.TransactionFactory; |
| import org.apache.phoenix.transaction.TransactionFactory.Provider; |
| import org.apache.phoenix.util.EncodedColumnsUtil; |
| import org.apache.phoenix.util.EnvironmentEdgeManager; |
| import org.apache.phoenix.util.IndexUtil; |
| import org.apache.phoenix.util.LogUtil; |
| import org.apache.phoenix.util.PhoenixKeyValueUtil; |
| import org.apache.phoenix.util.SQLCloseable; |
| import org.apache.phoenix.util.SchemaUtil; |
| import org.apache.phoenix.util.ServerUtil; |
| import org.apache.phoenix.util.SizedUtil; |
| import org.apache.phoenix.util.TransactionUtil; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.base.Preconditions; |
| import com.google.common.base.Predicate; |
| import com.google.common.collect.Iterators; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| |
| /** |
| * Tracks the uncommitted state |
| */ |
| public class MutationState implements SQLCloseable { |
| private static final Logger LOGGER = LoggerFactory.getLogger(MutationState.class); |
| private static final int[] EMPTY_STATEMENT_INDEX_ARRAY = new int[0]; |
| private static final int MAX_COMMIT_RETRIES = 3; |
| |
| private final PhoenixConnection connection; |
| private final long maxSize; |
| private final long maxSizeBytes; |
| private final long batchSize; |
| private final long batchSizeBytes; |
| private long batchCount = 0L; |
| private final Map<TableRef, MultiRowMutationState> mutations; |
| private final Set<String> uncommittedPhysicalNames = Sets.newHashSetWithExpectedSize(10); |
| |
| private long sizeOffset; |
| private int numRows = 0; |
| private long estimatedSize = 0; |
| private int[] uncommittedStatementIndexes = EMPTY_STATEMENT_INDEX_ARRAY; |
| private boolean isExternalTxContext = false; |
| private Map<TableRef, MultiRowMutationState> txMutations = Collections.emptyMap(); |
| |
| private PhoenixTransactionContext phoenixTransactionContext = PhoenixTransactionContext.NULL_CONTEXT; |
| |
| private final MutationMetricQueue mutationMetricQueue; |
| private ReadMetricQueue readMetricQueue; |
| |
| public MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection) { |
| this(maxSize, maxSizeBytes, connection, false, null); |
| } |
| |
| public MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, |
| PhoenixTransactionContext txContext) { |
| this(maxSize, maxSizeBytes, connection, false, txContext); |
| } |
| |
| public MutationState(MutationState mutationState) { |
| this(mutationState, mutationState.connection); |
| } |
| |
| public MutationState(MutationState mutationState, PhoenixConnection connection) { |
| this(mutationState.maxSize, mutationState.maxSizeBytes, connection, true, mutationState |
| .getPhoenixTransactionContext()); |
| } |
| |
| public MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, long sizeOffset) { |
| this(maxSize, maxSizeBytes, connection, false, null, sizeOffset); |
| } |
| |
| private MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, boolean subTask, |
| PhoenixTransactionContext txContext) { |
| this(maxSize, maxSizeBytes, connection, subTask, txContext, 0); |
| } |
| |
| private MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, boolean subTask, |
| PhoenixTransactionContext txContext, long sizeOffset) { |
| this(maxSize, maxSizeBytes, connection, Maps.<TableRef, MultiRowMutationState> newHashMapWithExpectedSize(5), |
| subTask, txContext); |
| this.sizeOffset = sizeOffset; |
| } |
| |
| MutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection, |
| Map<TableRef, MultiRowMutationState> mutations, boolean subTask, PhoenixTransactionContext txContext) { |
| this.maxSize = maxSize; |
| this.maxSizeBytes = maxSizeBytes; |
| this.connection = connection; |
| this.batchSize = connection.getMutateBatchSize(); |
| this.batchSizeBytes = connection.getMutateBatchSizeBytes(); |
| this.mutations = mutations; |
| boolean isMetricsEnabled = connection.isRequestLevelMetricsEnabled(); |
| this.mutationMetricQueue = isMetricsEnabled ? new MutationMetricQueue() |
| : NoOpMutationMetricsQueue.NO_OP_MUTATION_METRICS_QUEUE; |
| if (subTask) { |
| // this code path is only used while running child scans, we can't pass the txContext to child scans |
| // as it is not thread safe, so we use the tx member variable |
| phoenixTransactionContext = txContext.newTransactionContext(txContext, subTask); |
| } else if (txContext != null) { |
| isExternalTxContext = true; |
| phoenixTransactionContext = txContext.newTransactionContext(txContext, subTask); |
| } |
| } |
| |
| public MutationState(TableRef table, MultiRowMutationState mutations, long sizeOffset, long maxSize, |
| long maxSizeBytes, PhoenixConnection connection) throws SQLException { |
| this(maxSize, maxSizeBytes, connection, false, null, sizeOffset); |
| if (!mutations.isEmpty()) { |
| this.mutations.put(table, mutations); |
| } |
| this.numRows = mutations.size(); |
| this.estimatedSize = PhoenixKeyValueUtil.getEstimatedRowMutationSize(this.mutations); |
| throwIfTooBig(); |
| } |
| |
| public long getEstimatedSize() { |
| return estimatedSize; |
| } |
| |
| public long getMaxSize() { |
| return maxSize; |
| } |
| |
| public long getMaxSizeBytes() { |
| return maxSizeBytes; |
| } |
| |
| public PhoenixTransactionContext getPhoenixTransactionContext() { |
| return phoenixTransactionContext; |
| } |
| |
| /** |
| * Commit a write fence when creating an index so that we can detect when a data table transaction is started before |
| * the create index but completes after it. In this case, we need to rerun the data table transaction after the |
| * index creation so that the index rows are generated. See TEPHRA-157 for more information. |
| * |
| * @param dataTable |
| * the data table upon which an index is being added |
| * @throws SQLException |
| */ |
| public void commitDDLFence(PTable dataTable) throws SQLException { |
| if (dataTable.isTransactional()) { |
| try { |
| phoenixTransactionContext.commitDDLFence(dataTable); |
| } finally { |
| // The client expects a transaction to be in progress on the txContext while the |
| // VisibilityFence.prepareWait() starts a new tx and finishes/aborts it. After it's |
| // finished, we start a new one here. |
| // TODO: seems like an autonomous tx capability in Tephra would be useful here. |
| phoenixTransactionContext.begin(); |
| } |
| } |
| } |
| |
| public boolean checkpointIfNeccessary(MutationPlan plan) throws SQLException { |
| if (!phoenixTransactionContext.isTransactionRunning() || plan.getTargetRef() == null |
| || plan.getTargetRef().getTable() == null || !plan.getTargetRef().getTable().isTransactional()) { return false; } |
| Set<TableRef> sources = plan.getSourceRefs(); |
| if (sources.isEmpty()) { return false; } |
| // For a DELETE statement, we're always querying the table being deleted from. This isn't |
| // a problem, but it potentially could be if there are other references to the same table |
| // nested in the DELETE statement (as a sub query or join, for example). |
| TableRef ignoreForExcludeCurrent = plan.getOperation() == Operation.DELETE && sources.size() == 1 ? plan |
| .getTargetRef() : null; |
| boolean excludeCurrent = false; |
| String targetPhysicalName = plan.getTargetRef().getTable().getPhysicalName().getString(); |
| for (TableRef source : sources) { |
| if (source.getTable().isTransactional() && !source.equals(ignoreForExcludeCurrent)) { |
| String sourcePhysicalName = source.getTable().getPhysicalName().getString(); |
| if (targetPhysicalName.equals(sourcePhysicalName)) { |
| excludeCurrent = true; |
| break; |
| } |
| } |
| } |
| // If we're querying the same table we're updating, we must exclude our writes to |
| // it from being visible. |
| if (excludeCurrent) { |
| // If any source tables have uncommitted data prior to last checkpoint, |
| // then we must create a new checkpoint. |
| boolean hasUncommittedData = false; |
| for (TableRef source : sources) { |
| String sourcePhysicalName = source.getTable().getPhysicalName().getString(); |
| // Tracking uncommitted physical table names is an optimization that prevents us from |
| // having to do a checkpoint if no data has yet been written. If we're using an |
| // external transaction context, it's possible that data was already written at the |
| // current transaction timestamp, so we always checkpoint in that case is we're |
| // reading and writing to the same table. |
| if (source.getTable().isTransactional() |
| && (isExternalTxContext || uncommittedPhysicalNames.contains(sourcePhysicalName))) { |
| hasUncommittedData = true; |
| break; |
| } |
| } |
| |
| phoenixTransactionContext.checkpoint(hasUncommittedData); |
| |
| if (hasUncommittedData) { |
| uncommittedPhysicalNames.clear(); |
| } |
| return true; |
| } |
| return false; |
| } |
| |
| // Though MutationState is not thread safe in general, this method should be because it may |
| // be called by TableResultIterator in a multi-threaded manner. Since we do not want to expose |
| // the Transaction outside of MutationState, this seems reasonable, as the member variables |
| // would not change as these threads are running. We also clone mutationState to ensure that |
| // the transaction context won't change due to a commit when auto commit is true. |
| public Table getHTable(PTable table) throws SQLException { |
| Table htable = this.getConnection().getQueryServices().getTable(table.getPhysicalName().getBytes()); |
| if (table.isTransactional() && phoenixTransactionContext.isTransactionRunning()) { |
| // We're only using this table for reading, so we want it wrapped even if it's an index |
| htable = phoenixTransactionContext.getTransactionalTable(htable, table.isImmutableRows() || table.getType() == PTableType.INDEX); |
| } |
| return htable; |
| } |
| |
| public PhoenixConnection getConnection() { |
| return connection; |
| } |
| |
| public boolean isTransactionStarted() { |
| return phoenixTransactionContext.isTransactionRunning(); |
| } |
| |
| public long getInitialWritePointer() { |
| return phoenixTransactionContext.getTransactionId(); // First write pointer - won't change with checkpointing |
| } |
| |
| // For testing |
| public long getWritePointer() { |
| return phoenixTransactionContext.getWritePointer(); |
| } |
| |
| // For testing |
| public PhoenixVisibilityLevel getVisibilityLevel() { |
| return phoenixTransactionContext.getVisibilityLevel(); |
| } |
| |
| public boolean startTransaction(Provider provider) throws SQLException { |
| if (provider == null) { return false; } |
| if (!connection.getQueryServices().getProps() |
| .getBoolean(QueryServices.TRANSACTIONS_ENABLED, QueryServicesOptions.DEFAULT_TRANSACTIONS_ENABLED)) { throw new SQLExceptionInfo.Builder( |
| SQLExceptionCode.CANNOT_START_TXN_IF_TXN_DISABLED).build().buildException(); } |
| if (connection.getSCN() != null) { throw new SQLExceptionInfo.Builder( |
| SQLExceptionCode.CANNOT_START_TRANSACTION_WITH_SCN_SET).build().buildException(); } |
| |
| if (phoenixTransactionContext == PhoenixTransactionContext.NULL_CONTEXT) { |
| phoenixTransactionContext = provider.getTransactionProvider().getTransactionContext(connection); |
| } else { |
| if (provider != phoenixTransactionContext.getProvider()) { throw new SQLExceptionInfo.Builder( |
| SQLExceptionCode.CANNOT_MIX_TXN_PROVIDERS) |
| .setMessage(phoenixTransactionContext.getProvider().name() + " and " + provider.name()).build() |
| .buildException(); } |
| } |
| if (!isTransactionStarted()) { |
| // Clear any transactional state in case transaction was ended outside |
| // of Phoenix so we don't carry the old transaction state forward. We |
| // cannot call reset() here due to the case of having mutations and |
| // then transitioning from non transactional to transactional (which |
| // would end up clearing our uncommitted state). |
| resetTransactionalState(); |
| phoenixTransactionContext.begin(); |
| return true; |
| } |
| |
| return false; |
| } |
| |
| public static MutationState emptyMutationState(long maxSize, long maxSizeBytes, PhoenixConnection connection) { |
| MutationState state = new MutationState(maxSize, maxSizeBytes, connection, |
| Collections.<TableRef, MultiRowMutationState> emptyMap(), false, null); |
| state.sizeOffset = 0; |
| return state; |
| } |
| |
| private void throwIfTooBig() throws SQLException { |
| if (numRows > maxSize) { |
| resetState(); |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.MAX_MUTATION_SIZE_EXCEEDED).build().buildException(); |
| } |
| if (estimatedSize > maxSizeBytes) { |
| resetState(); |
| throw new SQLExceptionInfo.Builder(SQLExceptionCode.MAX_MUTATION_SIZE_BYTES_EXCEEDED).build() |
| .buildException(); |
| } |
| } |
| |
| public long getUpdateCount() { |
| return sizeOffset + numRows; |
| } |
| |
| private void joinMutationState(TableRef tableRef, MultiRowMutationState srcRows, |
| Map<TableRef, MultiRowMutationState> dstMutations) { |
| PTable table = tableRef.getTable(); |
| boolean isIndex = table.getType() == PTableType.INDEX; |
| boolean incrementRowCount = dstMutations == this.mutations; |
| MultiRowMutationState existingRows = dstMutations.put(tableRef, srcRows); |
| if (existingRows != null) { // Rows for that table already exist |
| // Loop through new rows and replace existing with new |
| for (Map.Entry<ImmutableBytesPtr, RowMutationState> rowEntry : srcRows.entrySet()) { |
| // Replace existing row with new row |
| RowMutationState existingRowMutationState = existingRows.put(rowEntry.getKey(), rowEntry.getValue()); |
| if (existingRowMutationState != null) { |
| Map<PColumn, byte[]> existingValues = existingRowMutationState.getColumnValues(); |
| if (existingValues != PRow.DELETE_MARKER) { |
| Map<PColumn, byte[]> newRow = rowEntry.getValue().getColumnValues(); |
| // if new row is PRow.DELETE_MARKER, it means delete, and we don't need to merge it with |
| // existing row. |
| if (newRow != PRow.DELETE_MARKER) { |
| // decrement estimated size by the size of the old row |
| estimatedSize -= existingRowMutationState.calculateEstimatedSize(); |
| // Merge existing column values with new column values |
| existingRowMutationState.join(rowEntry.getValue()); |
| // increment estimated size by the size of the new row |
| estimatedSize += existingRowMutationState.calculateEstimatedSize(); |
| // Now that the existing row has been merged with the new row, replace it back |
| // again (since it was merged with the new one above). |
| existingRows.put(rowEntry.getKey(), existingRowMutationState); |
| } |
| } |
| } else { |
| if (incrementRowCount && !isIndex) { // Don't count index rows in row count |
| numRows++; |
| // increment estimated size by the size of the new row |
| estimatedSize += rowEntry.getValue().calculateEstimatedSize(); |
| } |
| } |
| } |
| // Put the existing one back now that it's merged |
| dstMutations.put(tableRef, existingRows); |
| } else { |
| // Size new map at batch size as that's what it'll likely grow to. |
| MultiRowMutationState newRows = new MultiRowMutationState(connection.getMutateBatchSize()); |
| newRows.putAll(srcRows); |
| dstMutations.put(tableRef, newRows); |
| if (incrementRowCount && !isIndex) { |
| numRows += srcRows.size(); |
| // if we added all the rows from newMutationState we can just increment the |
| // estimatedSize by newMutationState.estimatedSize |
| estimatedSize += srcRows.estimatedSize; |
| } |
| } |
| } |
| |
| private void joinMutationState(Map<TableRef, MultiRowMutationState> srcMutations, |
| Map<TableRef, MultiRowMutationState> dstMutations) { |
| // Merge newMutation with this one, keeping state from newMutation for any overlaps |
| for (Map.Entry<TableRef, MultiRowMutationState> entry : srcMutations.entrySet()) { |
| // Replace existing entries for the table with new entries |
| TableRef tableRef = entry.getKey(); |
| MultiRowMutationState srcRows = entry.getValue(); |
| joinMutationState(tableRef, srcRows, dstMutations); |
| } |
| } |
| |
| /** |
| * Combine a newer mutation with this one, where in the event of overlaps, the newer one will take precedence. |
| * Combine any metrics collected for the newer mutation. |
| * |
| * @param newMutationState |
| * the newer mutation state |
| */ |
| public void join(MutationState newMutationState) throws SQLException { |
| if (this == newMutationState) { // Doesn't make sense |
| return; |
| } |
| |
| phoenixTransactionContext.join(newMutationState.getPhoenixTransactionContext()); |
| |
| this.sizeOffset += newMutationState.sizeOffset; |
| joinMutationState(newMutationState.mutations, this.mutations); |
| if (!newMutationState.txMutations.isEmpty()) { |
| if (txMutations.isEmpty()) { |
| txMutations = Maps.newHashMapWithExpectedSize(mutations.size()); |
| } |
| joinMutationState(newMutationState.txMutations, this.txMutations); |
| } |
| mutationMetricQueue.combineMetricQueues(newMutationState.mutationMetricQueue); |
| if (readMetricQueue == null) { |
| readMetricQueue = newMutationState.readMetricQueue; |
| } else if (readMetricQueue != null && newMutationState.readMetricQueue != null) { |
| readMetricQueue.combineReadMetrics(newMutationState.readMetricQueue); |
| } |
| throwIfTooBig(); |
| } |
| |
| private static ImmutableBytesPtr getNewRowKeyWithRowTimestamp(ImmutableBytesPtr ptr, long rowTimestamp, PTable table) { |
| RowKeySchema schema = table.getRowKeySchema(); |
| int rowTimestampColPos = table.getRowTimestampColPos(); |
| Field rowTimestampField = schema.getField(rowTimestampColPos); |
| byte[] rowTimestampBytes = rowTimestampField.getDataType() == PTimestamp.INSTANCE ? |
| PTimestamp.INSTANCE.toBytes(new Timestamp(rowTimestamp), rowTimestampField.getSortOrder()) : |
| PLong.INSTANCE.toBytes(rowTimestamp, rowTimestampField.getSortOrder()); |
| int oldOffset = ptr.getOffset(); |
| int oldLength = ptr.getLength(); |
| // Move the pointer to the start byte of the row timestamp pk |
| schema.position(ptr, 0, rowTimestampColPos); |
| byte[] b = ptr.get(); |
| int newOffset = ptr.getOffset(); |
| int length = ptr.getLength(); |
| for (int i = newOffset; i < newOffset + length; i++) { |
| // modify the underlying bytes array with the bytes of the row timestamp |
| b[i] = rowTimestampBytes[i - newOffset]; |
| } |
| // move the pointer back to where it was before. |
| ptr.set(ptr.get(), oldOffset, oldLength); |
| return ptr; |
| } |
| |
| private Iterator<Pair<PTable, List<Mutation>>> addRowMutations(final TableRef tableRef, |
| final MultiRowMutationState values, final long mutationTimestamp, final long serverTimestamp, |
| boolean includeAllIndexes, final boolean sendAll) { |
| final PTable table = tableRef.getTable(); |
| final List<PTable> indexList = includeAllIndexes ? |
| Lists.newArrayList(IndexMaintainer.maintainedIndexes(table.getIndexes().iterator())) : |
| IndexUtil.getClientMaintainedIndexes(table); |
| final Iterator<PTable> indexes = indexList.iterator(); |
| final List<Mutation> mutationList = Lists.newArrayListWithExpectedSize(values.size()); |
| final List<Mutation> mutationsPertainingToIndex = indexes.hasNext() ? Lists |
| .newArrayListWithExpectedSize(values.size()) : null; |
| generateMutations(tableRef, mutationTimestamp, serverTimestamp, values, mutationList, |
| mutationsPertainingToIndex); |
| return new Iterator<Pair<PTable, List<Mutation>>>() { |
| boolean isFirst = true; |
| Map<byte[], List<Mutation>> indexMutationsMap = null; |
| |
| @Override |
| public boolean hasNext() { |
| return isFirst || indexes.hasNext(); |
| } |
| |
| @Override |
| public Pair<PTable, List<Mutation>> next() { |
| if (isFirst) { |
| isFirst = false; |
| return new Pair<>(table, mutationList); |
| } |
| |
| PTable index = indexes.next(); |
| |
| List<Mutation> indexMutations = null; |
| try { |
| if (!mutationsPertainingToIndex.isEmpty()) { |
| if (table.isTransactional()) { |
| if (indexMutationsMap == null) { |
| PhoenixTxIndexMutationGenerator generator = PhoenixTxIndexMutationGenerator.newGenerator(connection, table, |
| indexList, mutationsPertainingToIndex.get(0).getAttributesMap()); |
| try (Table htable = connection.getQueryServices().getTable( |
| table.getPhysicalName().getBytes())) { |
| Collection<Pair<Mutation, byte[]>> allMutations = generator.getIndexUpdates(htable, |
| mutationsPertainingToIndex.iterator()); |
| indexMutationsMap = Maps.newTreeMap(Bytes.BYTES_COMPARATOR); |
| for (Pair<Mutation, byte[]> mutation : allMutations) { |
| List<Mutation> mutations = indexMutationsMap.get(mutation.getSecond()); |
| if (mutations == null) { |
| mutations = Lists.newArrayList(); |
| indexMutationsMap.put(mutation.getSecond(), mutations); |
| } |
| mutations.add(mutation.getFirst()); |
| } |
| } |
| } |
| indexMutations = indexMutationsMap.get(index.getPhysicalName().getBytes()); |
| } else { |
| indexMutations = IndexUtil.generateIndexData(table, index, values, |
| mutationsPertainingToIndex, connection.getKeyValueBuilder(), connection); |
| } |
| } |
| |
| // we may also have to include delete mutations for immutable tables if we are not processing all |
| // the tables in the mutations map |
| if (!sendAll) { |
| TableRef key = new TableRef(index); |
| MultiRowMutationState multiRowMutationState = mutations.remove(key); |
| if (multiRowMutationState != null) { |
| final List<Mutation> deleteMutations = Lists.newArrayList(); |
| generateMutations(key, mutationTimestamp, serverTimestamp, multiRowMutationState, deleteMutations, null); |
| if (indexMutations == null) { |
| indexMutations = deleteMutations; |
| } else { |
| indexMutations.addAll(deleteMutations); |
| } |
| } |
| } |
| } catch (SQLException | IOException e) { |
| throw new IllegalDataException(e); |
| } |
| return new Pair<PTable, List<Mutation>>(index, |
| indexMutations == null ? Collections.<Mutation> emptyList() |
| : indexMutations); |
| } |
| |
| @Override |
| public void remove() { |
| throw new UnsupportedOperationException(); |
| } |
| |
| }; |
| } |
| |
| private void generateMutations(final TableRef tableRef, final long mutationTimestamp, final long serverTimestamp, |
| final MultiRowMutationState values, final List<Mutation> mutationList, |
| final List<Mutation> mutationsPertainingToIndex) { |
| final PTable table = tableRef.getTable(); |
| boolean tableWithRowTimestampCol = table.getRowTimestampColPos() != -1; |
| Iterator<Map.Entry<ImmutableBytesPtr, RowMutationState>> iterator = values.entrySet().iterator(); |
| long timestampToUse = mutationTimestamp; |
| MultiRowMutationState modifiedValues = new MultiRowMutationState(16); |
| boolean wildcardIncludesDynamicCols = connection.getQueryServices().getProps().getBoolean( |
| WILDCARD_QUERY_DYNAMIC_COLS_ATTRIB, DEFAULT_WILDCARD_QUERY_DYNAMIC_COLS_ATTRIB); |
| while (iterator.hasNext()) { |
| Map.Entry<ImmutableBytesPtr, RowMutationState> rowEntry = iterator.next(); |
| byte[] onDupKeyBytes = rowEntry.getValue().getOnDupKeyBytes(); |
| boolean hasOnDupKey = onDupKeyBytes != null; |
| ImmutableBytesPtr key = rowEntry.getKey(); |
| RowMutationState state = rowEntry.getValue(); |
| if (tableWithRowTimestampCol) { |
| RowTimestampColInfo rowTsColInfo = state.getRowTimestampColInfo(); |
| if (rowTsColInfo.useServerTimestamp()) { |
| // regenerate the key with this timestamp. |
| key = getNewRowKeyWithRowTimestamp(key, serverTimestamp, table); |
| // since we are about to modify the byte[] stored in key (which changes its hashcode) |
| // we need to remove the entry from the values map and add a new entry with the modified byte[] |
| modifiedValues.put(key, state); |
| iterator.remove(); |
| timestampToUse = serverTimestamp; |
| } else { |
| if (rowTsColInfo.getTimestamp() != null) { |
| timestampToUse = rowTsColInfo.getTimestamp(); |
| } |
| } |
| } |
| PRow row = table.newRow(connection.getKeyValueBuilder(), timestampToUse, key, hasOnDupKey); |
| List<Mutation> rowMutations, rowMutationsPertainingToIndex; |
| if (rowEntry.getValue().getColumnValues() == PRow.DELETE_MARKER) { // means delete |
| row.delete(); |
| rowMutations = row.toRowMutations(); |
| // The DeleteCompiler already generates the deletes for indexes, so no need to do it again |
| rowMutationsPertainingToIndex = Collections.emptyList(); |
| } else { |
| for (Map.Entry<PColumn, byte[]> valueEntry : rowEntry.getValue().getColumnValues().entrySet()) { |
| row.setValue(valueEntry.getKey(), valueEntry.getValue()); |
| } |
| if (wildcardIncludesDynamicCols && row.setAttributesForDynamicColumnsIfReqd()) { |
| row.setAttributeToProcessDynamicColumnsMetadata(); |
| } |
| rowMutations = row.toRowMutations(); |
| // Pass through ON DUPLICATE KEY info through mutations |
| // In the case of the same clause being used on many statements, this will be |
| // inefficient because we're transmitting the same information for each mutation. |
| // TODO: use our ServerCache |
| for (Mutation mutation : rowMutations) { |
| if (onDupKeyBytes != null) { |
| mutation.setAttribute(PhoenixIndexBuilder.ATOMIC_OP_ATTRIB, onDupKeyBytes); |
| } |
| } |
| rowMutationsPertainingToIndex = rowMutations; |
| } |
| mutationList.addAll(rowMutations); |
| if (mutationsPertainingToIndex != null) mutationsPertainingToIndex.addAll(rowMutationsPertainingToIndex); |
| } |
| values.putAll(modifiedValues); |
| } |
| |
| /** |
| * Get the unsorted list of HBase mutations for the tables with uncommitted data. |
| * |
| * @return list of HBase mutations for uncommitted data. |
| */ |
| public Iterator<Pair<byte[], List<Mutation>>> toMutations(Long timestamp) { |
| return toMutations(false, timestamp); |
| } |
| |
| public Iterator<Pair<byte[], List<Mutation>>> toMutations() { |
| return toMutations(false, null); |
| } |
| |
| public Iterator<Pair<byte[], List<Mutation>>> toMutations(final boolean includeMutableIndexes) { |
| return toMutations(includeMutableIndexes, null); |
| } |
| |
| public Iterator<Pair<byte[], List<Mutation>>> toMutations(final boolean includeMutableIndexes, |
| final Long tableTimestamp) { |
| final Iterator<Map.Entry<TableRef, MultiRowMutationState>> iterator = this.mutations.entrySet().iterator(); |
| if (!iterator.hasNext()) { return Collections.emptyIterator(); } |
| Long scn = connection.getSCN(); |
| final long serverTimestamp = getTableTimestamp(tableTimestamp, scn); |
| final long mutationTimestamp = getMutationTimestamp(scn); |
| return new Iterator<Pair<byte[], List<Mutation>>>() { |
| private Map.Entry<TableRef, MultiRowMutationState> current = iterator.next(); |
| private Iterator<Pair<byte[], List<Mutation>>> innerIterator = init(); |
| |
| private Iterator<Pair<byte[], List<Mutation>>> init() { |
| final Iterator<Pair<PTable, List<Mutation>>> mutationIterator = |
| addRowMutations(current.getKey(), current.getValue(), |
| mutationTimestamp, serverTimestamp, includeMutableIndexes, true); |
| return new Iterator<Pair<byte[], List<Mutation>>>() { |
| @Override |
| public boolean hasNext() { |
| return mutationIterator.hasNext(); |
| } |
| |
| @Override |
| public Pair<byte[], List<Mutation>> next() { |
| Pair<PTable, List<Mutation>> pair = mutationIterator.next(); |
| return new Pair<byte[], List<Mutation>>(pair.getFirst().getPhysicalName() |
| .getBytes(), pair.getSecond()); |
| } |
| |
| @Override |
| public void remove() { |
| mutationIterator.remove(); |
| } |
| }; |
| } |
| |
| @Override |
| public boolean hasNext() { |
| return innerIterator.hasNext() || iterator.hasNext(); |
| } |
| |
| @Override |
| public Pair<byte[], List<Mutation>> next() { |
| if (!innerIterator.hasNext()) { |
| current = iterator.next(); |
| innerIterator = init(); |
| } |
| return innerIterator.next(); |
| } |
| |
| @Override |
| public void remove() { |
| throw new UnsupportedOperationException(); |
| } |
| |
| }; |
| } |
| |
| public static long getTableTimestamp(final Long tableTimestamp, Long scn) { |
| return (tableTimestamp != null && tableTimestamp != QueryConstants.UNSET_TIMESTAMP) ? tableTimestamp |
| : (scn == null ? HConstants.LATEST_TIMESTAMP : scn); |
| } |
| |
| public static long getMutationTimestamp(final Long scn) { |
| return scn == null ? HConstants.LATEST_TIMESTAMP : scn; |
| } |
| |
| /** |
| * Validates that the meta data is valid against the server meta data if we haven't yet done so. Otherwise, for |
| * every UPSERT VALUES call, we'd need to hit the server to see if the meta data has changed. |
| * |
| * @return the server time to use for the upsert |
| * @throws SQLException |
| * if the table or any columns no longer exist |
| */ |
| private long[] validateAll() throws SQLException { |
| int i = 0; |
| long[] timeStamps = new long[this.mutations.size()]; |
| for (Map.Entry<TableRef, MultiRowMutationState> entry : mutations.entrySet()) { |
| TableRef tableRef = entry.getKey(); |
| timeStamps[i++] = validateAndGetServerTimestamp(tableRef, entry.getValue()); |
| } |
| return timeStamps; |
| } |
| |
| private long validateAndGetServerTimestamp(TableRef tableRef, MultiRowMutationState rowKeyToColumnMap) |
| throws SQLException { |
| MetaDataClient client = new MetaDataClient(connection); |
| long serverTimeStamp = tableRef.getTimeStamp(); |
| // If we're auto committing, we've already validated the schema when we got the ColumnResolver, |
| // so no need to do it again here. |
| PTable table = tableRef.getTable(); |
| |
| // We generally don't re-resolve SYSTEM tables, but if it relies on ROW_TIMESTAMP, we must |
| // get the latest timestamp in order to upsert data with the correct server-side timestamp |
| // in case the ROW_TIMESTAMP is not provided in the UPSERT statement. |
| boolean hitServerForLatestTimestamp = |
| table.getRowTimestampColPos() != -1 && table.getType() == PTableType.SYSTEM; |
| MetaDataMutationResult result = client.updateCache(table.getSchemaName().getString(), |
| table.getTableName().getString(), hitServerForLatestTimestamp); |
| PTable resolvedTable = result.getTable(); |
| if (resolvedTable == null) { throw new TableNotFoundException(table.getSchemaName().getString(), table |
| .getTableName().getString()); } |
| // Always update tableRef table as the one we've cached may be out of date since when we executed |
| // the UPSERT VALUES call and updated in the cache before this. |
| tableRef.setTable(resolvedTable); |
| List<PTable> indexes = resolvedTable.getIndexes(); |
| for (PTable idxTtable : indexes) { |
| // If index is still active, but has a non zero INDEX_DISABLE_TIMESTAMP value, then infer that |
| // our failure mode is block writes on index failure. |
| if ((idxTtable.getIndexState() == PIndexState.ACTIVE || idxTtable.getIndexState() == PIndexState.PENDING_ACTIVE) |
| && idxTtable.getIndexDisableTimestamp() > 0) { throw new SQLExceptionInfo.Builder( |
| SQLExceptionCode.INDEX_FAILURE_BLOCK_WRITE).setSchemaName(table.getSchemaName().getString()) |
| .setTableName(table.getTableName().getString()).build().buildException(); } |
| } |
| long timestamp = result.getMutationTime(); |
| if (timestamp != QueryConstants.UNSET_TIMESTAMP) { |
| serverTimeStamp = timestamp; |
| if (result.wasUpdated()) { |
| List<PColumn> columns = Lists.newArrayListWithExpectedSize(table.getColumns().size()); |
| for (Map.Entry<ImmutableBytesPtr, RowMutationState> rowEntry : rowKeyToColumnMap.entrySet()) { |
| RowMutationState valueEntry = rowEntry.getValue(); |
| if (valueEntry != null) { |
| Map<PColumn, byte[]> colValues = valueEntry.getColumnValues(); |
| if (colValues != PRow.DELETE_MARKER) { |
| for (PColumn column : colValues.keySet()) { |
| if (!column.isDynamic()) columns.add(column); |
| } |
| } |
| } |
| } |
| for (PColumn column : columns) { |
| if (column != null) { |
| resolvedTable.getColumnFamily(column.getFamilyName().getString()).getPColumnForColumnName( |
| column.getName().getString()); |
| } |
| } |
| } |
| } |
| return serverTimeStamp == QueryConstants.UNSET_TIMESTAMP ? HConstants.LATEST_TIMESTAMP : serverTimeStamp; |
| } |
| |
| private static long calculateMutationSize(List<Mutation> mutations) { |
| long byteSize = 0; |
| if (GlobalClientMetrics.isMetricsEnabled()) { |
| for (Mutation mutation : mutations) { |
| byteSize += PhoenixKeyValueUtil.calculateMutationDiskSize(mutation); |
| } |
| } |
| GLOBAL_MUTATION_BYTES.update(byteSize); |
| return byteSize; |
| } |
| |
| public long getBatchSizeBytes() { |
| return batchSizeBytes; |
| } |
| |
| public long getBatchCount() { |
| return batchCount; |
| } |
| |
| private static class TableInfo { |
| |
| private final boolean isDataTable; |
| @Nonnull |
| private final PName hTableName; |
| @Nonnull |
| private final TableRef origTableRef; |
| private final PTable pTable; |
| |
| public TableInfo(boolean isDataTable, PName hTableName, TableRef origTableRef, PTable pTable) { |
| super(); |
| checkNotNull(hTableName); |
| checkNotNull(origTableRef); |
| this.isDataTable = isDataTable; |
| this.hTableName = hTableName; |
| this.origTableRef = origTableRef; |
| this.pTable = pTable; |
| } |
| |
| public boolean isDataTable() { |
| return isDataTable; |
| } |
| |
| public PName getHTableName() { |
| return hTableName; |
| } |
| |
| public TableRef getOrigTableRef() { |
| return origTableRef; |
| } |
| |
| public PTable getPTable() { |
| return pTable; |
| } |
| |
| @Override |
| public int hashCode() { |
| final int prime = 31; |
| int result = 1; |
| result = prime * result + hTableName.hashCode(); |
| result = prime * result + (isDataTable ? 1231 : 1237); |
| return result; |
| } |
| |
| @Override |
| public boolean equals(Object obj) { |
| if (this == obj) return true; |
| if (obj == null) return false; |
| if (getClass() != obj.getClass()) return false; |
| TableInfo other = (TableInfo)obj; |
| if (!hTableName.equals(other.hTableName)) return false; |
| if (isDataTable != other.isDataTable) return false; |
| if (!pTable.equals(other.pTable)) return false; |
| |
| return true; |
| } |
| |
| } |
| private void send(Iterator<TableRef> tableRefIterator) throws SQLException { |
| int i = 0; |
| long[] serverTimeStamps = null; |
| boolean sendAll = false; |
| if (tableRefIterator == null) { |
| serverTimeStamps = validateAll(); |
| tableRefIterator = mutations.keySet().iterator(); |
| sendAll = true; |
| } |
| |
| MultiRowMutationState multiRowMutationState; |
| Map<TableInfo, List<Mutation>> physicalTableMutationMap = Maps.newLinkedHashMap(); |
| // add tracing for this operation |
| try (TraceScope trace = Tracing.startNewSpan(connection, "Committing mutations to tables")) { |
| Span span = trace.getSpan(); |
| ImmutableBytesWritable indexMetaDataPtr = new ImmutableBytesWritable(); |
| while (tableRefIterator.hasNext()) { |
| // at this point we are going through mutations for each table |
| final TableRef tableRef = tableRefIterator.next(); |
| multiRowMutationState = mutations.get(tableRef); |
| if (multiRowMutationState == null || multiRowMutationState.isEmpty()) { |
| continue; |
| } |
| // Validate as we go if transactional since we can undo if a problem occurs (which is unlikely) |
| long |
| serverTimestamp = |
| serverTimeStamps == null ? |
| validateAndGetServerTimestamp(tableRef, multiRowMutationState) : |
| serverTimeStamps[i++]; |
| final PTable table = tableRef.getTable(); |
| Long scn = connection.getSCN(); |
| long mutationTimestamp = scn == null ? |
| (table.isTransactional() == true ? HConstants.LATEST_TIMESTAMP : EnvironmentEdgeManager.currentTimeMillis()) |
| : scn; |
| Iterator<Pair<PTable, List<Mutation>>> |
| mutationsIterator = |
| addRowMutations(tableRef, multiRowMutationState, mutationTimestamp, |
| serverTimestamp, false, sendAll); |
| // build map from physical table to mutation list |
| boolean isDataTable = true; |
| while (mutationsIterator.hasNext()) { |
| Pair<PTable, List<Mutation>> pair = mutationsIterator.next(); |
| PTable logicalTable = pair.getFirst(); |
| List<Mutation> mutationList = pair.getSecond(); |
| |
| TableInfo tableInfo = new TableInfo(isDataTable, logicalTable.getPhysicalName(), |
| tableRef, logicalTable); |
| |
| List<Mutation> |
| oldMutationList = |
| physicalTableMutationMap.put(tableInfo, mutationList); |
| if (oldMutationList != null) mutationList.addAll(0, oldMutationList); |
| isDataTable = false; |
| } |
| // For transactions, track the statement indexes as we send data |
| // over because our CommitException should include all statements |
| // involved in the transaction since none of them would have been |
| // committed in the event of a failure. |
| if (table.isTransactional()) { |
| addUncommittedStatementIndexes(multiRowMutationState.values()); |
| if (txMutations.isEmpty()) { |
| txMutations = Maps.newHashMapWithExpectedSize(mutations.size()); |
| } |
| // Keep all mutations we've encountered until a commit or rollback. |
| // This is not ideal, but there's not good way to get the values back |
| // in the event that we need to replay the commit. |
| // Copy TableRef so we have the original PTable and know when the |
| // indexes have changed. |
| joinMutationState(new TableRef(tableRef), multiRowMutationState, txMutations); |
| } |
| } |
| |
| Map<TableInfo, List<Mutation>> unverifiedIndexMutations = new LinkedHashMap<>(); |
| Map<TableInfo, List<Mutation>> verifiedOrDeletedIndexMutations = new LinkedHashMap<>(); |
| filterIndexCheckerMutations(physicalTableMutationMap, unverifiedIndexMutations, |
| verifiedOrDeletedIndexMutations); |
| |
| // Phase 1: Send index mutations with the empty column value = "unverified" |
| sendMutations(unverifiedIndexMutations.entrySet().iterator(), span, indexMetaDataPtr); |
| |
| // Phase 2: Send data table and other indexes |
| sendMutations(physicalTableMutationMap.entrySet().iterator(), span, indexMetaDataPtr); |
| |
| // Phase 3: Send put index mutations with the empty column value = "verified" and/or delete index mutations |
| try { |
| sendMutations(verifiedOrDeletedIndexMutations.entrySet().iterator(), span, indexMetaDataPtr); |
| } catch (SQLException ex) { |
| // TODO: add a metric here |
| LOGGER.warn( |
| "Ignoring exception that happened during setting index verified value to verified=TRUE " |
| + verifiedOrDeletedIndexMutations.toString(), |
| ex); |
| } |
| } |
| } |
| |
| private void sendMutations(Iterator<Entry<TableInfo, List<Mutation>>> mutationsIterator, Span span, ImmutableBytesWritable indexMetaDataPtr) |
| throws SQLException { |
| while (mutationsIterator.hasNext()) { |
| Entry<TableInfo, List<Mutation>> pair = mutationsIterator.next(); |
| TableInfo tableInfo = pair.getKey(); |
| byte[] htableName = tableInfo.getHTableName().getBytes(); |
| List<Mutation> mutationList = pair.getValue(); |
| List<List<Mutation>> mutationBatchList = |
| getMutationBatchList(batchSize, batchSizeBytes, mutationList); |
| |
| // create a span per target table |
| // TODO maybe we can be smarter about the table name to string here? |
| Span child = Tracing.child(span, "Writing mutation batch for table: " + Bytes.toString(htableName)); |
| |
| int retryCount = 0; |
| boolean shouldRetry = false; |
| long numMutations = 0; |
| long mutationSizeBytes = 0; |
| long mutationCommitTime = 0; |
| long numFailedMutations = 0; |
| |
| long startTime = 0; |
| boolean shouldRetryIndexedMutation = false; |
| IndexWriteException iwe = null; |
| do { |
| TableRef origTableRef = tableInfo.getOrigTableRef(); |
| PTable table = origTableRef.getTable(); |
| table.getIndexMaintainers(indexMetaDataPtr, connection); |
| final ServerCache cache = tableInfo.isDataTable() ? |
| IndexMetaDataCacheClient.setMetaDataOnMutations(connection, table, |
| mutationList, indexMetaDataPtr) : null; |
| // If we haven't retried yet, retry for this case only, as it's possible that |
| // a split will occur after we send the index metadata cache to all known |
| // region servers. |
| shouldRetry = cache != null; |
| SQLException sqlE = null; |
| Table hTable = connection.getQueryServices().getTable(htableName); |
| try { |
| if (table.isTransactional()) { |
| // Track tables to which we've sent uncommitted data |
| if (tableInfo.isDataTable()) { |
| uncommittedPhysicalNames.add(table.getPhysicalName().getString()); |
| phoenixTransactionContext.markDMLFence(table); |
| } |
| // Only pass true for last argument if the index is being written to on it's own (i.e. initial |
| // index population), not if it's being written to for normal maintenance due to writes to |
| // the data table. This case is different because the initial index population does not need |
| // to be done transactionally since the index is only made active after all writes have |
| // occurred successfully. |
| hTable = phoenixTransactionContext.getTransactionalTableWriter(connection, table, hTable, tableInfo.isDataTable() && table.getType() == PTableType.INDEX); |
| } |
| numMutations = mutationList.size(); |
| GLOBAL_MUTATION_BATCH_SIZE.update(numMutations); |
| mutationSizeBytes = calculateMutationSize(mutationList); |
| |
| startTime = System.currentTimeMillis(); |
| child.addTimelineAnnotation("Attempt " + retryCount); |
| Iterator<List<Mutation>> itrListMutation = mutationBatchList.iterator(); |
| while (itrListMutation.hasNext()) { |
| final List<Mutation> mutationBatch = itrListMutation.next(); |
| if (shouldRetryIndexedMutation) { |
| // if there was an index write failure, retry the mutation in a loop |
| final Table finalHTable = hTable; |
| final ImmutableBytesWritable finalindexMetaDataPtr = |
| indexMetaDataPtr; |
| final PTable finalPTable = table; |
| PhoenixIndexFailurePolicy.doBatchWithRetries(new MutateCommand() { |
| @Override |
| public void doMutation() throws IOException { |
| try { |
| finalHTable.batch(mutationBatch, null); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| throw new IOException(e); |
| } catch (IOException e) { |
| e = updateTableRegionCacheIfNecessary(e); |
| throw e; |
| } |
| } |
| |
| @Override |
| public List<Mutation> getMutationList() { |
| return mutationBatch; |
| } |
| |
| private IOException |
| updateTableRegionCacheIfNecessary(IOException ioe) { |
| SQLException sqlE = |
| ServerUtil.parseLocalOrRemoteServerException(ioe); |
| if (sqlE != null |
| && sqlE.getErrorCode() == SQLExceptionCode.INDEX_METADATA_NOT_FOUND |
| .getErrorCode()) { |
| try { |
| connection.getQueryServices().clearTableRegionCache( |
| finalHTable.getName()); |
| IndexMetaDataCacheClient.setMetaDataOnMutations( |
| connection, finalPTable, mutationBatch, |
| finalindexMetaDataPtr); |
| } catch (SQLException e) { |
| return ServerUtil.createIOException( |
| "Exception during updating index meta data cache", |
| ioe); |
| } |
| } |
| return ioe; |
| } |
| }, iwe, connection, connection.getQueryServices().getProps()); |
| shouldRetryIndexedMutation = false; |
| } else { |
| hTable.batch(mutationBatch, null); |
| } |
| // remove each batch from the list once it gets applied |
| // so when failures happens for any batch we only start |
| // from that batch only instead of doing duplicate reply of already |
| // applied batches from entire list, also we can set |
| // REPLAY_ONLY_INDEX_WRITES for first batch |
| // only in case of 1121 SQLException |
| itrListMutation.remove(); |
| |
| batchCount++; |
| if (LOGGER.isDebugEnabled()) |
| LOGGER.debug("Sent batch of " + mutationBatch.size() + " for " |
| + Bytes.toString(htableName)); |
| } |
| child.stop(); |
| child.stop(); |
| shouldRetry = false; |
| mutationCommitTime = System.currentTimeMillis() - startTime; |
| GLOBAL_MUTATION_COMMIT_TIME.update(mutationCommitTime); |
| numFailedMutations = 0; |
| |
| // Remove batches as we process them |
| mutations.remove(origTableRef); |
| if (tableInfo.isDataTable()) { |
| numRows -= numMutations; |
| // recalculate the estimated size |
| estimatedSize = PhoenixKeyValueUtil.getEstimatedRowMutationSize(mutations); |
| } |
| } catch (Exception e) { |
| mutationCommitTime = System.currentTimeMillis() - startTime; |
| long serverTimestamp = ServerUtil.parseServerTimestamp(e); |
| SQLException inferredE = ServerUtil.parseServerExceptionOrNull(e); |
| if (inferredE != null) { |
| if (shouldRetry |
| && retryCount == 0 |
| && inferredE.getErrorCode() == SQLExceptionCode.INDEX_METADATA_NOT_FOUND |
| .getErrorCode()) { |
| // Swallow this exception once, as it's possible that we split after sending the index |
| // metadata |
| // and one of the region servers doesn't have it. This will cause it to have it the next |
| // go around. |
| // If it fails again, we don't retry. |
| String msg = "Swallowing exception and retrying after clearing meta cache on connection. " |
| + inferredE; |
| LOGGER.warn(LogUtil.addCustomAnnotations(msg, connection)); |
| connection.getQueryServices().clearTableRegionCache(TableName.valueOf(htableName)); |
| |
| // add a new child span as this one failed |
| child.addTimelineAnnotation(msg); |
| child.stop(); |
| child = Tracing.child(span, "Failed batch, attempting retry"); |
| |
| continue; |
| } else if (inferredE.getErrorCode() == SQLExceptionCode.INDEX_WRITE_FAILURE.getErrorCode()) { |
| iwe = PhoenixIndexFailurePolicy.getIndexWriteException(inferredE); |
| if (iwe != null && !shouldRetryIndexedMutation) { |
| // For an index write failure, the data table write succeeded, |
| // so when we retry we need to set REPLAY_WRITES |
| // for first batch in list only. |
| for (Mutation m : mutationBatchList.get(0)) { |
| if (!PhoenixIndexMetaData.isIndexRebuild( |
| m.getAttributesMap())){ |
| m.setAttribute(BaseScannerRegionObserver.REPLAY_WRITES, |
| BaseScannerRegionObserver.REPLAY_ONLY_INDEX_WRITES |
| ); |
| } |
| PhoenixKeyValueUtil.setTimestamp(m, serverTimestamp); |
| } |
| shouldRetry = true; |
| shouldRetryIndexedMutation = true; |
| continue; |
| } |
| } |
| e = inferredE; |
| } |
| // Throw to client an exception that indicates the statements that |
| // were not committed successfully. |
| int[] uncommittedStatementIndexes = getUncommittedStatementIndexes(); |
| sqlE = new CommitException(e, uncommittedStatementIndexes, serverTimestamp); |
| numFailedMutations = uncommittedStatementIndexes.length; |
| GLOBAL_MUTATION_BATCH_FAILED_COUNT.update(numFailedMutations); |
| } finally { |
| MutationMetric mutationsMetric = new MutationMetric(numMutations, mutationSizeBytes, |
| mutationCommitTime, numFailedMutations); |
| mutationMetricQueue.addMetricsForTable(Bytes.toString(htableName), mutationsMetric); |
| try { |
| if (cache != null) cache.close(); |
| } finally { |
| try { |
| hTable.close(); |
| } catch (IOException e) { |
| if (sqlE != null) { |
| sqlE.setNextException(ServerUtil.parseServerException(e)); |
| } else { |
| sqlE = ServerUtil.parseServerException(e); |
| } |
| } |
| if (sqlE != null) { throw sqlE; } |
| } |
| } |
| } while (shouldRetry && retryCount++ < 1); |
| } |
| } |
| |
| private void filterIndexCheckerMutations(Map<TableInfo, List<Mutation>> mutationMap, |
| Map<TableInfo, List<Mutation>> unverifiedIndexMutations, |
| Map<TableInfo, List<Mutation>> verifiedOrDeletedIndexMutations) throws SQLException { |
| Iterator<Entry<TableInfo, List<Mutation>>> mapIter = mutationMap.entrySet().iterator(); |
| |
| while (mapIter.hasNext()) { |
| Entry<TableInfo, List<Mutation>> pair = mapIter.next(); |
| TableInfo tableInfo = pair.getKey(); |
| PTable logicalTable = tableInfo.getPTable(); |
| if (tableInfo.getOrigTableRef().getTable().isImmutableRows() |
| && IndexUtil.isGlobalIndexCheckerEnabled(connection, |
| tableInfo.getHTableName())) { |
| |
| byte[] emptyCF = SchemaUtil.getEmptyColumnFamily(logicalTable); |
| byte[] emptyCQ = EncodedColumnsUtil.getEmptyKeyValueInfo(logicalTable).getFirst(); |
| List<Mutation> mutations = pair.getValue(); |
| |
| for (Mutation m : mutations) { |
| if (m == null) { |
| continue; |
| } |
| if (m instanceof Delete) { |
| Put put = new Put(m.getRow()); |
| put.addColumn(emptyCF, emptyCQ, IndexRegionObserver.getMaxTimestamp(m), |
| IndexRegionObserver.UNVERIFIED_BYTES); |
| // The Delete gets marked as unverified in Phase 1 and gets deleted on Phase 3. |
| addToMap(unverifiedIndexMutations, tableInfo, put); |
| addToMap(verifiedOrDeletedIndexMutations, tableInfo, m); |
| } else if (m instanceof Put) { |
| long timestamp = IndexRegionObserver.getMaxTimestamp(m); |
| |
| // Phase 1 index mutations are set to unverified |
| // Send entire mutation with the unverified status |
| // Remove the empty column prepared by Index codec as we need to change its value |
| IndexRegionObserver.removeEmptyColumn(m, emptyCF, emptyCQ); |
| ((Put) m).addColumn(emptyCF, emptyCQ, timestamp, IndexRegionObserver.UNVERIFIED_BYTES); |
| addToMap(unverifiedIndexMutations, tableInfo, m); |
| |
| // Phase 3 mutations are verified |
| Put verifiedPut = new Put(m.getRow()); |
| verifiedPut.addColumn(emptyCF, emptyCQ, timestamp, IndexRegionObserver.VERIFIED_BYTES); |
| addToMap(verifiedOrDeletedIndexMutations, tableInfo, verifiedPut); |
| } else { |
| addToMap(unverifiedIndexMutations, tableInfo, m); |
| } |
| } |
| |
| mapIter.remove(); |
| } |
| |
| } |
| } |
| |
| private void addToMap(Map<TableInfo, List<Mutation>> map, TableInfo tableInfo, Mutation mutation) { |
| List<Mutation> mutations = null; |
| if (map.containsKey(tableInfo)) { |
| mutations = map.get(tableInfo); |
| } else { |
| mutations = Lists.newArrayList(); |
| } |
| mutations.add(mutation); |
| map.put(tableInfo, mutations); |
| } |
| |
| /** |
| * |
| * Split the list of mutations into multiple lists. since a single row update can contain multiple mutations, |
| * we only check if the current batch has exceeded the row or size limit for different rows, |
| * so that mutations for a single row don't end up in different batches. |
| * |
| * @param allMutationList |
| * List of HBase mutations |
| * @return List of lists of mutations |
| */ |
| public static List<List<Mutation>> getMutationBatchList(long batchSize, long batchSizeBytes, List<Mutation> allMutationList) { |
| Preconditions.checkArgument(batchSize> 1, |
| "Mutation types are put or delete, for one row all mutations must be in one batch."); |
| Preconditions.checkArgument(batchSizeBytes > 0, "Batch size must be larger than 0"); |
| List<List<Mutation>> mutationBatchList = Lists.newArrayList(); |
| List<Mutation> currentList = Lists.newArrayList(); |
| List<Mutation> sameRowList = Lists.newArrayList(); |
| long currentBatchSizeBytes = 0L; |
| for (int i = 0; i < allMutationList.size(); ) { |
| long sameRowBatchSize = 1L; |
| Mutation mutation = allMutationList.get(i); |
| long sameRowMutationSizeBytes = PhoenixKeyValueUtil.calculateMutationDiskSize(mutation); |
| sameRowList.add(mutation); |
| while (i + 1 < allMutationList.size() && |
| Bytes.compareTo(allMutationList.get(i + 1).getRow(), mutation.getRow()) == 0) { |
| Mutation sameRowMutation = allMutationList.get(i + 1); |
| sameRowList.add(sameRowMutation); |
| sameRowMutationSizeBytes += PhoenixKeyValueUtil.calculateMutationDiskSize(sameRowMutation); |
| sameRowBatchSize++; |
| i++; |
| } |
| |
| if (currentList.size() + sameRowBatchSize > batchSize || |
| currentBatchSizeBytes + sameRowMutationSizeBytes > batchSizeBytes) { |
| if (currentList.size() > 0) { |
| mutationBatchList.add(currentList); |
| currentList = Lists.newArrayList(); |
| currentBatchSizeBytes = 0L; |
| } |
| } |
| |
| currentList.addAll(sameRowList); |
| currentBatchSizeBytes += sameRowMutationSizeBytes; |
| sameRowList.clear(); |
| i++; |
| } |
| |
| if (currentList.size() > 0) { |
| mutationBatchList.add(currentList); |
| } |
| return mutationBatchList; |
| } |
| |
| public byte[] encodeTransaction() throws SQLException { |
| return phoenixTransactionContext.encodeTransaction(); |
| } |
| |
| private void addUncommittedStatementIndexes(Collection<RowMutationState> rowMutations) { |
| for (RowMutationState rowMutationState : rowMutations) { |
| uncommittedStatementIndexes = joinSortedIntArrays(uncommittedStatementIndexes, |
| rowMutationState.getStatementIndexes()); |
| } |
| } |
| |
| private int[] getUncommittedStatementIndexes() { |
| for (MultiRowMutationState rowMutationMap : mutations.values()) { |
| addUncommittedStatementIndexes(rowMutationMap.values()); |
| } |
| return uncommittedStatementIndexes; |
| } |
| |
| @Override |
| public void close() throws SQLException {} |
| |
| private void resetState() { |
| numRows = 0; |
| estimatedSize = 0; |
| this.mutations.clear(); |
| phoenixTransactionContext = PhoenixTransactionContext.NULL_CONTEXT; |
| } |
| |
| private void resetTransactionalState() { |
| phoenixTransactionContext.reset(); |
| txMutations = Collections.emptyMap(); |
| uncommittedPhysicalNames.clear(); |
| uncommittedStatementIndexes = EMPTY_STATEMENT_INDEX_ARRAY; |
| } |
| |
| public void rollback() throws SQLException { |
| try { |
| phoenixTransactionContext.abort(); |
| } finally { |
| resetState(); |
| } |
| } |
| |
| public void commit() throws SQLException { |
| Map<TableRef, MultiRowMutationState> txMutations = Collections.emptyMap(); |
| int retryCount = 0; |
| do { |
| boolean sendSuccessful = false; |
| boolean retryCommit = false; |
| SQLException sqlE = null; |
| try { |
| send(); |
| txMutations = this.txMutations; |
| sendSuccessful = true; |
| } catch (SQLException e) { |
| sqlE = e; |
| } finally { |
| try { |
| boolean finishSuccessful = false; |
| try { |
| if (sendSuccessful) { |
| phoenixTransactionContext.commit(); |
| finishSuccessful = true; |
| } |
| } catch (SQLException e) { |
| if (LOGGER.isInfoEnabled()) |
| LOGGER.info(e.getClass().getName() + " at timestamp " + getInitialWritePointer() |
| + " with retry count of " + retryCount); |
| retryCommit = (e.getErrorCode() == SQLExceptionCode.TRANSACTION_CONFLICT_EXCEPTION |
| .getErrorCode() && retryCount < MAX_COMMIT_RETRIES); |
| if (sqlE == null) { |
| sqlE = e; |
| } else { |
| sqlE.setNextException(e); |
| } |
| } finally { |
| // If send fails or finish fails, abort the tx |
| if (!finishSuccessful) { |
| try { |
| phoenixTransactionContext.abort(); |
| if (LOGGER.isInfoEnabled()) LOGGER.info("Abort successful"); |
| } catch (SQLException e) { |
| if (LOGGER.isInfoEnabled()) LOGGER.info("Abort failed with " + e); |
| if (sqlE == null) { |
| sqlE = e; |
| } else { |
| sqlE.setNextException(e); |
| } |
| } |
| } |
| } |
| } finally { |
| TransactionFactory.Provider provider = phoenixTransactionContext.getProvider(); |
| try { |
| resetState(); |
| } finally { |
| if (retryCommit) { |
| startTransaction(provider); |
| // Add back read fences |
| Set<TableRef> txTableRefs = txMutations.keySet(); |
| for (TableRef tableRef : txTableRefs) { |
| PTable dataTable = tableRef.getTable(); |
| phoenixTransactionContext.markDMLFence(dataTable); |
| } |
| try { |
| // Only retry if an index was added |
| retryCommit = shouldResubmitTransaction(txTableRefs); |
| } catch (SQLException e) { |
| retryCommit = false; |
| if (sqlE == null) { |
| sqlE = e; |
| } else { |
| sqlE.setNextException(e); |
| } |
| } |
| } |
| if (sqlE != null && !retryCommit) { throw sqlE; } |
| } |
| } |
| } |
| // Retry commit once if conflict occurred and index was added |
| if (!retryCommit) { |
| break; |
| } |
| retryCount++; |
| mutations.putAll(txMutations); |
| } while (true); |
| } |
| |
| /** |
| * Determines whether indexes were added to mutated tables while the transaction was in progress. |
| * |
| * @return true if indexes were added and false otherwise. |
| * @throws SQLException |
| */ |
| private boolean shouldResubmitTransaction(Set<TableRef> txTableRefs) throws SQLException { |
| if (LOGGER.isInfoEnabled()) LOGGER.info("Checking for index updates as of " + getInitialWritePointer()); |
| MetaDataClient client = new MetaDataClient(connection); |
| PMetaData cache = connection.getMetaDataCache(); |
| boolean addedAnyIndexes = false; |
| boolean allImmutableTables = !txTableRefs.isEmpty(); |
| for (TableRef tableRef : txTableRefs) { |
| PTable dataTable = tableRef.getTable(); |
| List<PTable> oldIndexes; |
| PTableRef ptableRef = cache.getTableRef(dataTable.getKey()); |
| oldIndexes = ptableRef.getTable().getIndexes(); |
| // Always check at server for metadata change, as it's possible that the table is configured to not check |
| // for metadata changes |
| // but in this case, the tx manager is telling us it's likely that there has been a change. |
| MetaDataMutationResult result = client.updateCache(dataTable.getTenantId(), dataTable.getSchemaName() |
| .getString(), dataTable.getTableName().getString(), true); |
| long timestamp = TransactionUtil.getResolvedTime(connection, result); |
| tableRef.setTimeStamp(timestamp); |
| PTable updatedDataTable = result.getTable(); |
| if (updatedDataTable == null) { throw new TableNotFoundException(dataTable.getSchemaName().getString(), |
| dataTable.getTableName().getString()); } |
| allImmutableTables &= updatedDataTable.isImmutableRows(); |
| tableRef.setTable(updatedDataTable); |
| if (!addedAnyIndexes) { |
| // TODO: in theory we should do a deep equals check here, as it's possible |
| // that an index was dropped and recreated with the same name but different |
| // indexed/covered columns. |
| addedAnyIndexes = (!oldIndexes.equals(updatedDataTable.getIndexes())); |
| if (LOGGER.isInfoEnabled()) |
| LOGGER.info((addedAnyIndexes ? "Updates " : "No updates ") + "as of " + timestamp + " to " |
| + updatedDataTable.getName().getString() + " with indexes " + updatedDataTable.getIndexes()); |
| } |
| } |
| if (LOGGER.isInfoEnabled()) |
| LOGGER.info((addedAnyIndexes ? "Updates " : "No updates ") + "to indexes as of " + getInitialWritePointer() |
| + " over " + (allImmutableTables ? " all immutable tables" : " some mutable tables")); |
| // If all tables are immutable, we know the conflict we got was due to our DDL/DML fence. |
| // If any indexes were added, then the conflict might be due to DDL/DML fence. |
| return allImmutableTables || addedAnyIndexes; |
| } |
| |
| /** |
| * Send to HBase any uncommitted data for transactional tables. |
| * |
| * @return true if any data was sent and false otherwise. |
| * @throws SQLException |
| */ |
| public boolean sendUncommitted() throws SQLException { |
| return sendUncommitted(mutations.keySet().iterator()); |
| } |
| |
| /** |
| * Support read-your-own-write semantics by sending uncommitted data to HBase prior to running a query. In this way, |
| * they are visible to subsequent reads but are not actually committed until commit is called. |
| * |
| * @param tableRefs |
| * @return true if any data was sent and false otherwise. |
| * @throws SQLException |
| */ |
| public boolean sendUncommitted(Iterator<TableRef> tableRefs) throws SQLException { |
| |
| if (phoenixTransactionContext.isTransactionRunning()) { |
| // Initialize visibility so that transactions see their own writes. |
| // The checkpoint() method will set it to not see writes if necessary. |
| phoenixTransactionContext.setVisibilityLevel(PhoenixVisibilityLevel.SNAPSHOT); |
| } |
| |
| Iterator<TableRef> filteredTableRefs = Iterators.filter(tableRefs, new Predicate<TableRef>() { |
| @Override |
| public boolean apply(TableRef tableRef) { |
| return tableRef.getTable().isTransactional(); |
| } |
| }); |
| if (filteredTableRefs.hasNext()) { |
| // FIXME: strip table alias to prevent equality check from failing due to alias mismatch on null alias. |
| // We really should be keying the tables based on the physical table name. |
| List<TableRef> strippedAliases = Lists.newArrayListWithExpectedSize(mutations.keySet().size()); |
| while (filteredTableRefs.hasNext()) { |
| TableRef tableRef = filteredTableRefs.next(); |
| // REVIEW: unclear if we need this given we start transactions when resolving a table |
| if (tableRef.getTable().isTransactional()) { |
| startTransaction(tableRef.getTable().getTransactionProvider()); |
| } |
| strippedAliases.add(new TableRef(null, tableRef.getTable(), tableRef.getTimeStamp(), tableRef |
| .getLowerBoundTimeStamp(), tableRef.hasDynamicCols())); |
| } |
| send(strippedAliases.iterator()); |
| return true; |
| } |
| return false; |
| } |
| |
| public void send() throws SQLException { |
| send(null); |
| } |
| |
| public static int[] joinSortedIntArrays(int[] a, int[] b) { |
| int[] result = new int[a.length + b.length]; |
| int i = 0, j = 0, k = 0, current; |
| while (i < a.length && j < b.length) { |
| current = a[i] < b[j] ? a[i++] : b[j++]; |
| for (; i < a.length && a[i] == current; i++) |
| ; |
| for (; j < b.length && b[j] == current; j++) |
| ; |
| result[k++] = current; |
| } |
| while (i < a.length) { |
| for (current = a[i++]; i < a.length && a[i] == current; i++) |
| ; |
| result[k++] = current; |
| } |
| while (j < b.length) { |
| for (current = b[j++]; j < b.length && b[j] == current; j++) |
| ; |
| result[k++] = current; |
| } |
| return Arrays.copyOf(result, k); |
| } |
| |
| @Immutable |
| public static class RowTimestampColInfo { |
| private final boolean useServerTimestamp; |
| private final Long rowTimestamp; |
| |
| public static final RowTimestampColInfo NULL_ROWTIMESTAMP_INFO = new RowTimestampColInfo(false, null); |
| |
| public RowTimestampColInfo(boolean autoGenerate, Long value) { |
| this.useServerTimestamp = autoGenerate; |
| this.rowTimestamp = value; |
| } |
| |
| public boolean useServerTimestamp() { |
| return useServerTimestamp; |
| } |
| |
| public Long getTimestamp() { |
| return rowTimestamp; |
| } |
| } |
| |
| public static class MultiRowMutationState { |
| private Map<ImmutableBytesPtr, RowMutationState> rowKeyToRowMutationState; |
| private long estimatedSize; |
| |
| public MultiRowMutationState(int size) { |
| this.rowKeyToRowMutationState = Maps.newHashMapWithExpectedSize(size); |
| this.estimatedSize = 0; |
| } |
| |
| public RowMutationState put(ImmutableBytesPtr ptr, RowMutationState rowMutationState) { |
| estimatedSize += rowMutationState.calculateEstimatedSize(); |
| return rowKeyToRowMutationState.put(ptr, rowMutationState); |
| } |
| |
| public void putAll(MultiRowMutationState other) { |
| estimatedSize += other.estimatedSize; |
| rowKeyToRowMutationState.putAll(other.rowKeyToRowMutationState); |
| } |
| |
| public boolean isEmpty() { |
| return rowKeyToRowMutationState.isEmpty(); |
| } |
| |
| public int size() { |
| return rowKeyToRowMutationState.size(); |
| } |
| |
| public Set<Entry<ImmutableBytesPtr, RowMutationState>> entrySet() { |
| return rowKeyToRowMutationState.entrySet(); |
| } |
| |
| public void clear() { |
| rowKeyToRowMutationState.clear(); |
| estimatedSize = 0; |
| } |
| |
| public Collection<RowMutationState> values() { |
| return rowKeyToRowMutationState.values(); |
| } |
| } |
| |
| public static class RowMutationState { |
| @Nonnull |
| private Map<PColumn, byte[]> columnValues; |
| private int[] statementIndexes; |
| @Nonnull |
| private final RowTimestampColInfo rowTsColInfo; |
| private byte[] onDupKeyBytes; |
| private long colValuesSize; |
| |
| public RowMutationState(@Nonnull Map<PColumn, byte[]> columnValues, long colValuesSize, int statementIndex, |
| @Nonnull RowTimestampColInfo rowTsColInfo, byte[] onDupKeyBytes) { |
| checkNotNull(columnValues); |
| checkNotNull(rowTsColInfo); |
| this.columnValues = columnValues; |
| this.statementIndexes = new int[] { statementIndex }; |
| this.rowTsColInfo = rowTsColInfo; |
| this.onDupKeyBytes = onDupKeyBytes; |
| this.colValuesSize = colValuesSize; |
| } |
| |
| public long calculateEstimatedSize() { |
| return colValuesSize + statementIndexes.length * SizedUtil.INT_SIZE + SizedUtil.LONG_SIZE |
| + (onDupKeyBytes != null ? onDupKeyBytes.length : 0); |
| } |
| |
| byte[] getOnDupKeyBytes() { |
| return onDupKeyBytes; |
| } |
| |
| public Map<PColumn, byte[]> getColumnValues() { |
| return columnValues; |
| } |
| |
| int[] getStatementIndexes() { |
| return statementIndexes; |
| } |
| |
| void join(RowMutationState newRow) { |
| // If we already have a row and the new row has an ON DUPLICATE KEY clause |
| // ignore the new values (as that's what the server will do). |
| if (newRow.onDupKeyBytes == null) { |
| // increment the column value size by the new row column value size |
| colValuesSize += newRow.colValuesSize; |
| for (Map.Entry<PColumn, byte[]> entry : newRow.columnValues.entrySet()) { |
| PColumn col = entry.getKey(); |
| byte[] oldValue = columnValues.put(col, entry.getValue()); |
| if (oldValue != null) { |
| // decrement column value size by the size of all column values that were replaced |
| colValuesSize -= (col.getEstimatedSize() + oldValue.length); |
| } |
| } |
| } |
| // Concatenate ON DUPLICATE KEY bytes to allow multiple |
| // increments of the same row in the same commit batch. |
| this.onDupKeyBytes = PhoenixIndexBuilder.combineOnDupKey(this.onDupKeyBytes, newRow.onDupKeyBytes); |
| statementIndexes = joinSortedIntArrays(statementIndexes, newRow.getStatementIndexes()); |
| } |
| |
| @Nonnull |
| RowTimestampColInfo getRowTimestampColInfo() { |
| return rowTsColInfo; |
| } |
| |
| } |
| |
| public ReadMetricQueue getReadMetricQueue() { |
| return readMetricQueue; |
| } |
| |
| public void setReadMetricQueue(ReadMetricQueue readMetricQueue) { |
| this.readMetricQueue = readMetricQueue; |
| } |
| |
| public MutationMetricQueue getMutationMetricQueue() { |
| return mutationMetricQueue; |
| } |
| |
| } |
