You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by sk...@apache.org on 2019/12/12 18:23:37 UTC
[phoenix] branch master updated: PHOENIX-5590 IndexTool always runs
with direct
This is an automated email from the ASF dual-hosted git repository.
skadam pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new 4581437 PHOENIX-5590 IndexTool always runs with direct
4581437 is described below
commit 4581437262e60969ab995b5f6d28b924a630dfae
Author: Gokcen Iskender <gi...@salesforce.com>
AuthorDate: Tue Nov 26 14:31:06 2019 -0800
PHOENIX-5590 IndexTool always runs with direct
Signed-off-by: s.kadam <s....@apache.org>
---
.../end2end/IndexToolForPartialBuildIT.java | 1 +
.../org/apache/phoenix/end2end/IndexToolIT.java | 23 +--
.../coprocessor/tasks/IndexRebuildTask.java | 2 +-
.../apache/phoenix/mapreduce/index/IndexTool.java | 101 +++---------
.../mapreduce/index/PhoenixIndexImportMapper.java | 177 ---------------------
5 files changed, 30 insertions(+), 274 deletions(-)
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java
index 50515b1..3027554 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolForPartialBuildIT.java
@@ -237,6 +237,7 @@ public class IndexToolForPartialBuildIT extends BaseOwnClusterIT {
// complete index rebuild
args.add("-it");
args.add(indexName);
+ args.add("-runfg");
args.add("-op");
args.add("/tmp/output/partialTable_");
return args.toArray(new String[0]);
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
index de13833..0eae120 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/IndexToolIT.java
@@ -38,11 +38,8 @@ import java.util.Map;
import java.util.Properties;
import java.util.UUID;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
@@ -50,12 +47,10 @@ import org.apache.hadoop.hbase.regionserver.HRegion;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
-import org.apache.phoenix.end2end.index.GlobalIndexCheckerIT;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.index.IndexTool;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.mapreduce.index.PhoenixIndexImportDirectMapper;
-import org.apache.phoenix.mapreduce.index.PhoenixIndexImportMapper;
import org.apache.phoenix.mapreduce.index.PhoenixServerBuildIndexMapper;
import org.apache.phoenix.query.QueryServices;
@@ -658,10 +653,9 @@ public class IndexToolIT extends BaseUniqueNamesOwnClusterIT {
args.add(indxTable);
if (directApi) {
args.add("-direct");
- // Need to run this job in foreground for the test to be deterministic
- args.add("-runfg");
}
-
+ // Need to run this job in foreground for the test to be deterministic
+ args.add("-runfg");
if (useSnapshot) {
args.add("-snap");
}
@@ -702,15 +696,10 @@ public class IndexToolIT extends BaseUniqueNamesOwnClusterIT {
boolean transactional = dataTable.isTransactional();
boolean localIndex = PTable.IndexType.LOCAL.equals(indexTable.getIndexType());
- if (directApi) {
- if ((localIndex || !transactional) && !useSnapshot) {
- assertEquals(job.getMapperClass(), PhoenixServerBuildIndexMapper.class);
- } else {
- assertEquals(job.getMapperClass(), PhoenixIndexImportDirectMapper.class);
- }
- }
- else {
- assertEquals(job.getMapperClass(), PhoenixIndexImportMapper.class);
+ if ((localIndex || !transactional) && !useSnapshot) {
+ assertEquals(job.getMapperClass(), PhoenixServerBuildIndexMapper.class);
+ } else {
+ assertEquals(job.getMapperClass(), PhoenixIndexImportDirectMapper.class);
}
}
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/tasks/IndexRebuildTask.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/tasks/IndexRebuildTask.java
index 168605f..91d7e67 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/tasks/IndexRebuildTask.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/tasks/IndexRebuildTask.java
@@ -94,7 +94,7 @@ public class IndexRebuildTask extends BaseTask {
// Run index tool async.
boolean runForeground = false;
Map.Entry<Integer, Job> indexToolRes = IndexTool
- .run(conf, taskRecord.getSchemaName(), taskRecord.getTableName(), indexName, true,
+ .run(conf, taskRecord.getSchemaName(), taskRecord.getTableName(), indexName,
false, taskRecord.getTenantId(), shouldDisable, rebuildAll, runForeground);
int status = indexToolRes.getKey();
if (status != 0) {
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
index 1961fe3..d01ba64 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
@@ -51,7 +51,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Admin;
@@ -59,8 +58,6 @@ import org.apache.hadoop.hbase.client.RegionLocator;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
-import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
@@ -125,7 +122,6 @@ public class IndexTool extends Configured implements Tool {
private boolean isPartialBuild;
private String qDataTable;
private String qIndexTable;
- private boolean useDirectApi;
private boolean useSnapshot;
private boolean isLocalIndexBuild;
private boolean shouldDeleteBeforeRebuild;
@@ -147,7 +143,7 @@ public class IndexTool extends Configured implements Tool {
"To build indexes for a data table from least disabledTimeStamp");
private static final Option DIRECT_API_OPTION = new Option("direct", "direct", false,
- "If specified, we avoid the bulk load (optional)");
+ "This parameter is deprecated. Direct mode will be used whether it is set or not. Keeping it for backwards compatibility.");
private static final double DEFAULT_SPLIT_SAMPLING_RATE = 10.0;
@@ -233,27 +229,20 @@ public class IndexTool extends Configured implements Tool {
throw new IllegalStateException(DATA_TABLE_OPTION.getLongOpt() + " is a mandatory "
+ "parameter");
}
-
- if (!(cmdLine.hasOption(PARTIAL_REBUILD_OPTION.getOpt()) || cmdLine.hasOption(DIRECT_API_OPTION.getOpt()))
- && !cmdLine.hasOption(OUTPUT_PATH_OPTION.getOpt())) {
- throw new IllegalStateException(OUTPUT_PATH_OPTION.getLongOpt() + " is a mandatory " + "parameter");
- }
if (cmdLine.hasOption(PARTIAL_REBUILD_OPTION.getOpt()) && cmdLine.hasOption(INDEX_TABLE_OPTION.getOpt())) {
throw new IllegalStateException("Index name should not be passed with " + PARTIAL_REBUILD_OPTION.getLongOpt());
}
+ if (!cmdLine.hasOption(PARTIAL_REBUILD_OPTION.getOpt()) && !cmdLine.hasOption(INDEX_TABLE_OPTION.getOpt())) {
+ throw new IllegalStateException("Index name should be passed unless it is a partial rebuild.");
+ }
+
if (cmdLine.hasOption(PARTIAL_REBUILD_OPTION.getOpt()) && cmdLine.hasOption(DELETE_ALL_AND_REBUILD_OPTION.getOpt())) {
throw new IllegalStateException(DELETE_ALL_AND_REBUILD_OPTION.getLongOpt() + " is not compatible with "
+ PARTIAL_REBUILD_OPTION.getLongOpt());
}
-
- if (!(cmdLine.hasOption(DIRECT_API_OPTION.getOpt())) && cmdLine.hasOption(INDEX_TABLE_OPTION.getOpt())
- && cmdLine.hasOption(RUN_FOREGROUND_OPTION
- .getOpt())) {
- throw new IllegalStateException(RUN_FOREGROUND_OPTION.getLongOpt()
- + " is applicable only for " + DIRECT_API_OPTION.getLongOpt());
- }
+
boolean splitIndex = cmdLine.hasOption(AUTO_SPLIT_INDEX_OPTION.getOpt()) || cmdLine.hasOption(SPLIT_INDEX_OPTION.getOpt());
if (splitIndex && !cmdLine.hasOption(INDEX_TABLE_OPTION.getOpt())) {
throw new IllegalStateException("Must pass an index name for the split index option");
@@ -305,10 +294,10 @@ public class IndexTool extends Configured implements Tool {
Long.toString(TransactionUtil.convertToNanoseconds(maxTimeRange)));
configuration.set(PhoenixConfigurationUtil.TX_PROVIDER, pDataTable.getTransactionProvider().name());
}
- if (useSnapshot || !useDirectApi || (!isLocalIndexBuild && pDataTable.isTransactional())) {
+ if (useSnapshot || (!isLocalIndexBuild && pDataTable.isTransactional())) {
configuration.set(PhoenixConfigurationUtil.CURRENT_SCN_VALUE,
Long.toString(maxTimeRange));
- return configureJobForAysncIndex();
+ return configureJobForAsyncIndex();
} else {
// Local and non-transactional global indexes to be built on the server side
// It is safe not to set CURRENT_SCN_VALUE for server side rebuilds, in order to make sure that
@@ -429,10 +418,7 @@ public class IndexTool extends Configured implements Tool {
}
- private Job configureJobForAysncIndex()
-
- throws Exception {
-
+ private Job configureJobForAsyncIndex() throws Exception {
String physicalIndexTable = pIndexTable.getPhysicalName().getString();
final PhoenixConnection pConnection = connection.unwrap(PhoenixConnection.class);
final PostIndexDDLCompiler ddlCompiler =
@@ -456,13 +442,17 @@ public class IndexTool extends Configured implements Tool {
PhoenixRuntime.generateColumnInfo(connection, qIndexTable, indexColumns);
ColumnInfoToStringEncoderDecoder.encode(configuration, columnMetadataList);
- fs = outputPath.getFileSystem(configuration);
- fs.delete(outputPath, true);
+ if (outputPath != null) {
+ fs = outputPath.getFileSystem(configuration);
+ fs.delete(outputPath, true);
+ }
final String jobName = String.format(INDEX_JOB_NAME_TEMPLATE, schemaName, dataTable, indexTable);
final Job job = Job.getInstance(configuration, jobName);
job.setJarByClass(IndexTool.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
- FileOutputFormat.setOutputPath(job, outputPath);
+ if (outputPath != null) {
+ FileOutputFormat.setOutputPath(job, outputPath);
+ }
if (!useSnapshot) {
PhoenixMapReduceUtil.setInput(job, PhoenixIndexDBWritable.class, qDataTable, selectQuery);
@@ -490,16 +480,11 @@ public class IndexTool extends Configured implements Tool {
}
TableMapReduceUtil.initCredentials(job);
- if (useDirectApi) {
- job.setMapperClass(PhoenixIndexImportDirectMapper.class);
- return configureSubmittableJobUsingDirectApi(job);
- } else {
- return configureRunnableJobUsingBulkLoad(job, outputPath);
- }
+ job.setMapperClass(PhoenixIndexImportDirectMapper.class);
+ return configureSubmittableJobUsingDirectApi(job);
}
- private Job configureJobForServerBuildIndex()
- throws Exception {
+ private Job configureJobForServerBuildIndex() throws Exception {
long indexRebuildQueryTimeoutMs =
configuration.getLong(QueryServices.INDEX_REBUILD_QUERY_TIMEOUT_ATTRIB,
QueryServicesOptions.DEFAULT_INDEX_REBUILD_QUERY_TIMEOUT);
@@ -555,29 +540,6 @@ public class IndexTool extends Configured implements Tool {
}
/**
- * Submits the job and waits for completion.
- * @param job
- * @param outputPath
- * @return
- * @throws Exception
- */
- private Job configureRunnableJobUsingBulkLoad(Job job, Path outputPath) throws Exception {
- job.setMapperClass(PhoenixIndexImportMapper.class);
- job.setMapOutputKeyClass(ImmutableBytesWritable.class);
- job.setMapOutputValueClass(KeyValue.class);
- final Configuration configuration = job.getConfiguration();
- final String physicalIndexTable =
- PhoenixConfigurationUtil.getPhysicalTableName(configuration);
- try(org.apache.hadoop.hbase.client.Connection conn =
- ConnectionFactory.createConnection(configuration)) {
- TableName tablename = TableName.valueOf(physicalIndexTable);
- HFileOutputFormat2.configureIncrementalLoad(job, conn.getTable(tablename),
- conn.getRegionLocator(tablename));
- }
- return job;
- }
-
- /**
* Uses the HBase Front Door Api to write to index table. Submits the job and either returns or
* waits for the job completion based on runForeground parameter.
*
@@ -588,9 +550,7 @@ public class IndexTool extends Configured implements Tool {
* @return
* @throws Exception
*/
- private Job configureSubmittableJobUsingDirectApi(Job job)
- throws Exception {
-
+ private Job configureSubmittableJobUsingDirectApi(Job job) throws Exception {
job.setReducerClass(PhoenixIndexImportDirectReducer.class);
Configuration conf = job.getConfiguration();
HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
@@ -643,7 +603,6 @@ public class IndexTool extends Configured implements Tool {
try(Connection tempConn = ConnectionUtil.getInputConnection(configuration)) {
pDataTable = PhoenixRuntime.getTableNoCache(tempConn, qDataTable);
}
- useDirectApi = cmdLine.hasOption(DIRECT_API_OPTION.getOpt());
String basePath=cmdLine.getOptionValue(OUTPUT_PATH_OPTION.getOpt());
boolean isForeground = cmdLine.hasOption(RUN_FOREGROUND_OPTION.getOpt());
useSnapshot = cmdLine.hasOption(SNAPSHOT_OPTION.getOpt());
@@ -708,7 +667,7 @@ public class IndexTool extends Configured implements Tool {
jobFactory = new JobFactory(connection, configuration, outputPath);
job = jobFactory.getJob();
- if (!isForeground && useDirectApi) {
+ if (!isForeground) {
LOGGER.info("Running Index Build in Background - Submit async and exit");
job.submit();
return 0;
@@ -718,19 +677,6 @@ public class IndexTool extends Configured implements Tool {
boolean result = job.waitForCompletion(true);
if (result) {
- if (!useDirectApi && indexTable != null) {
- if (isLocalIndexBuild) {
- validateSplitForLocalIndex(splitKeysBeforeJob, regionLocator);
- }
- LOGGER.info("Loading HFiles from {}", outputPath);
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(configuration);
- loader.doBulkLoad(outputPath, connection.unwrap(PhoenixConnection.class)
- .getQueryServices().getAdmin(), htable, regionLocator);
- htable.close();
- // Without direct API, we need to update the index state to ACTIVE from client.
- IndexToolUtil.updateIndexState(connection, qDataTable, indexTable, PIndexState.ACTIVE);
- fs.delete(outputPath, true);
- }
return 0;
} else {
LOGGER.error("IndexTool job failed! Check logs for errors..");
@@ -950,7 +896,7 @@ public class IndexTool extends Configured implements Tool {
}
public static Map.Entry<Integer, Job> run(Configuration conf, String schemaName, String dataTable, String indexTable,
- boolean directApi, boolean useSnapshot, String tenantId, boolean disableBefore, boolean shouldDeleteBeforeRebuild, boolean runForeground) throws Exception {
+ boolean useSnapshot, String tenantId, boolean disableBefore, boolean shouldDeleteBeforeRebuild, boolean runForeground) throws Exception {
final List<String> args = Lists.newArrayList();
if (schemaName != null) {
args.add("-s");
@@ -960,9 +906,6 @@ public class IndexTool extends Configured implements Tool {
args.add(dataTable);
args.add("-it");
args.add(indexTable);
- if (directApi) {
- args.add("-direct");
- }
if (runForeground) {
args.add("-runfg");
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportMapper.java
deleted file mode 100644
index 74cc3dd..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportMapper.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/*
- * 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.mapreduce.index;
-
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Properties;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.Pair;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.mapreduce.ImportPreUpsertKeyValueProcessor;
-import org.apache.phoenix.mapreduce.PhoenixJobCounters;
-import org.apache.phoenix.mapreduce.util.ConnectionUtil;
-import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
-import org.apache.phoenix.transaction.PhoenixTransactionProvider;
-import org.apache.phoenix.transaction.TransactionFactory;
-import org.apache.phoenix.util.ColumnInfo;
-import org.apache.phoenix.util.PhoenixKeyValueUtil;
-import org.apache.phoenix.util.PhoenixRuntime;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Lists;
-
-/**
- * Mapper that hands over rows from data table to the index table.
- *
- */
-public class PhoenixIndexImportMapper extends Mapper<NullWritable, PhoenixIndexDBWritable, ImmutableBytesWritable, KeyValue> {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(PhoenixIndexImportMapper.class);
-
- private final PhoenixIndexDBWritable indxWritable = new PhoenixIndexDBWritable();
-
- private List<ColumnInfo> indxTblColumnMetadata ;
-
- private Connection connection;
-
- private String indexTableName;
-
- private ImportPreUpsertKeyValueProcessor preUpdateProcessor;
-
- private PreparedStatement pStatement;
-
- @Override
- protected void setup(final Context context) throws IOException, InterruptedException {
- super.setup(context);
- final Configuration configuration = context.getConfiguration();
- try {
- indxTblColumnMetadata = PhoenixConfigurationUtil.getUpsertColumnMetadataList(context.getConfiguration());
- indxWritable.setColumnMetadata(indxTblColumnMetadata);
-
- preUpdateProcessor = PhoenixConfigurationUtil.loadPreUpsertProcessor(configuration);
- indexTableName = PhoenixConfigurationUtil.getPhysicalTableName(configuration);
- final Properties overrideProps = new Properties ();
- String scn = configuration.get(PhoenixConfigurationUtil.CURRENT_SCN_VALUE);
- String txScnValue = configuration.get(PhoenixConfigurationUtil.TX_SCN_VALUE);
- if(txScnValue==null) {
- overrideProps.put(PhoenixRuntime.BUILD_INDEX_AT_ATTRIB, scn);
- }
- connection = ConnectionUtil.getOutputConnection(configuration,overrideProps);
- connection.setAutoCommit(false);
- final String upsertQuery = PhoenixConfigurationUtil.getUpsertStatement(configuration);
- this.pStatement = connection.prepareStatement(upsertQuery);
-
- } catch (SQLException e) {
- tryClosingConnection();
- throw new RuntimeException(e.getMessage());
- }
- }
-
- @Override
- protected void map(NullWritable key, PhoenixIndexDBWritable record, Context context)
- throws IOException, InterruptedException {
-
- context.getCounter(PhoenixJobCounters.INPUT_RECORDS).increment(1);
-
- PhoenixTransactionProvider provider = null;
- Configuration conf = context.getConfiguration();
- long ts = HConstants.LATEST_TIMESTAMP;
- String txnIdStr = conf.get(PhoenixConfigurationUtil.TX_SCN_VALUE);
- if (txnIdStr != null) {
- ts = Long.parseLong(txnIdStr);
- provider = TransactionFactory.Provider.getDefault().getTransactionProvider();
- String txnProviderStr = conf.get(PhoenixConfigurationUtil.TX_PROVIDER);
- if (txnProviderStr != null) {
- provider = TransactionFactory.Provider.valueOf(txnProviderStr).getTransactionProvider();
- }
- }
- try {
- final ImmutableBytesWritable outputKey = new ImmutableBytesWritable();
- final List<Object> values = record.getValues();
- indxWritable.setValues(values);
- indxWritable.write(this.pStatement);
- this.pStatement.execute();
-
- PhoenixConnection pconn = connection.unwrap(PhoenixConnection.class);
- final Iterator<Pair<byte[],List<Mutation>>> iterator = pconn.getMutationState().toMutations(true);
- while (iterator.hasNext()) {
- Pair<byte[], List<Mutation>> pair = iterator.next();
- if (Bytes.compareTo(Bytes.toBytes(indexTableName), pair.getFirst()) != 0) {
- // skip edits for other tables
- continue;
- }
- List<Cell> keyValues = Lists.newArrayListWithExpectedSize(pair.getSecond().size() * 5); // Guess-timate 5 key values per row
- for (Mutation mutation : pair.getSecond()) {
- if (mutation instanceof Put) {
- if (provider != null) {
- mutation = provider.markPutAsCommitted((Put)mutation, ts, ts);
- }
- for (List<Cell> cellList : mutation.getFamilyCellMap().values()) {
- List<Cell>keyValueList = preUpdateProcessor.preUpsert(mutation.getRow(), cellList);
- keyValues.addAll(keyValueList);
- }
- }
- }
- keyValues.sort(pconn.getKeyValueBuilder().getKeyValueComparator());
- for (Cell kv : keyValues) {
- outputKey.set(kv.getRowArray(), kv.getRowOffset(), kv.getRowLength());
- context.write(outputKey, PhoenixKeyValueUtil.maybeCopyCell(kv));
- }
- context.getCounter(PhoenixJobCounters.OUTPUT_RECORDS).increment(1);
- }
- connection.rollback();
- } catch (SQLException e) {
- LOGGER.error("Error {} while read/write of a record ",e.getMessage());
- context.getCounter(PhoenixJobCounters.FAILED_RECORDS).increment(1);
- throw new RuntimeException(e);
- }
- }
-
- @Override
- protected void cleanup(Context context) throws IOException, InterruptedException {
- super.cleanup(context);
- tryClosingConnection();
- }
-
- private void tryClosingConnection() {
- if (connection != null) {
- try {
- connection.close();
- } catch (SQLException e) {
- LOGGER.error("Error while closing connection in the PhoenixIndexMapper class ", e);
- }
- }
- }
-}