Skip to content

PHOENIX-6153 Table Map Reduce job after a Snapshot based job fails wi…#902

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
sakshamgangwar wants to merge 1 commit into from
Closed

PHOENIX-6153 Table Map Reduce job after a Snapshot based job fails wi… #902

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
162 changes: 142 additions & 20 deletions phoenix-core/src/it/java/org/apache/phoenix/end2end/TableSnapshotReadsMapReduceIT.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -23,6 +23,7 @@
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.sql.Array;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
Expand All@@ -40,14 +41,22 @@
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.phoenix.iterate.TestingMapReduceParallelScanGrouper;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.PhoenixOutputFormat;
import org.apache.phoenix.mapreduce.PhoenixTestingInputFormat;
import org.apache.phoenix.mapreduce.index.PhoenixIndexDBWritable;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;
import org.apache.phoenix.schema.types.PDouble;
import org.apache.phoenix.schema.types.PhoenixArray;
import org.apache.phoenix.util.ReadOnlyProps;
import org.junit.Assert;
import org.junit.Before;
Expand All@@ -62,14 +71,25 @@ public class TableSnapshotReadsMapReduceIT extends BaseUniqueNamesOwnClusterIT {

private static final Logger LOGGER = LoggerFactory.getLogger(TableSnapshotReadsMapReduceIT.class);

private static final String STOCK_NAME = "STOCK_NAME";
private static final String RECORDING_YEAR = "RECORDING_YEAR";
private static final String RECORDINGS_QUARTER = "RECORDINGS_QUARTER";
private static final String MAX_RECORDING = "MAX_RECORDING";
private final static String SNAPSHOT_NAME = "FOO";
private static final String FIELD1 = "FIELD1";
private static final String FIELD2 = "FIELD2";
private static final String FIELD3 = "FIELD3";
private String CREATE_TABLE = "CREATE TABLE IF NOT EXISTS %s ( " +
" FIELD1 VARCHAR NOT NULL , FIELD2 VARCHAR , FIELD3 INTEGER CONSTRAINT pk PRIMARY KEY (FIELD1 ))";
private String UPSERT = "UPSERT into %s values (?, ?, ?)";

private static final String CREATE_STOCK_TABLE =
"CREATE TABLE IF NOT EXISTS %s ( " + STOCK_NAME + " VARCHAR NOT NULL , " + RECORDING_YEAR
+ " INTEGER NOT NULL, " + RECORDINGS_QUARTER + " "
+ " DOUBLE array[] CONSTRAINT pk PRIMARY KEY ( " + STOCK_NAME + ", "
+ RECORDING_YEAR + " )) " + "SPLIT ON ('AA')";
private static final String CREATE_STOCK_STATS_TABLE =
"CREATE TABLE IF NOT EXISTS %s(" + STOCK_NAME + " VARCHAR NOT NULL , " + MAX_RECORDING
+ " DOUBLE CONSTRAINT pk PRIMARY KEY (" + STOCK_NAME + " ))";
private static List<List<Object>> result;
private long timestamp;
private String tableName;
Expand All@@ -86,10 +106,11 @@ public static synchronized void doSetup() throws Exception {
@Before
public void before() throws SQLException, IOException {
// create table
Connection conn = DriverManager.getConnection(getUrl());
tableName = generateUniqueName();
conn.createStatement().execute(String.format(CREATE_TABLE, tableName));
conn.commit();
try (Connection conn = DriverManager.getConnection(getUrl())) {
tableName = generateUniqueName();
conn.createStatement().execute(String.format(CREATE_TABLE, tableName));
conn.commit();
}

// configure Phoenix M/R job to read snapshot
conf = getUtility().getConfiguration();
Expand DownExpand Up@@ -126,6 +147,90 @@ public void testMapReduceSnapshotWithLimit() throws Exception {
configureJob(job, tableName, inputQuery, null, false);
}

@Test
public void testSnapshotMapReduceJobNotImpactingTableMapReduceJob() throws Exception {
//Submitting and asserting successful Map Reduce Job over snapshots
PhoenixMapReduceUtil
.setInput(job, PhoenixIndexDBWritable.class, SNAPSHOT_NAME, tableName, tmpDir, null,
FIELD1, FIELD2, FIELD3);
configureJob(job, tableName, null, null, false);

//Asserting that snapshot name is set in configuration
Configuration config = job.getConfiguration();
Assert.assertEquals("Correct snapshot name not found in configuration", SNAPSHOT_NAME,
config.get(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY));

try (Connection conn = DriverManager.getConnection(getUrl())) {
// create table
tableName = generateUniqueName();
conn.createStatement().execute(String.format(CREATE_TABLE, tableName));
conn.commit();

//Submitting next map reduce job over table and making sure that it does not fail with
// any wrong snapshot properties set in common configurations which are
// used across all jobs.
job = createAndTestJob(conn);
}
//Asserting that snapshot name is no more set in common shared configuration
config = job.getConfiguration();
Assert.assertNull("Snapshot name is not null in Configuration",
config.get(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY));

}

private Job createAndTestJob(Connection conn)
throws SQLException, IOException, InterruptedException, ClassNotFoundException {
String stockTableName = generateUniqueName();
String stockStatsTableName = generateUniqueName();
conn.createStatement().execute(String.format(CREATE_STOCK_TABLE, stockTableName));
conn.createStatement().execute(String.format(CREATE_STOCK_STATS_TABLE, stockStatsTableName));
conn.commit();
final Configuration conf = ((PhoenixConnection) conn).getQueryServices().getConfiguration();
Job job = Job.getInstance(conf);
PhoenixMapReduceUtil.setInput(job, MapReduceIT.StockWritable.class, PhoenixTestingInputFormat.class,
stockTableName, null, STOCK_NAME, RECORDING_YEAR, "0." + RECORDINGS_QUARTER);
testJob(conn, job, stockTableName, stockStatsTableName);
return job;
}

private void testJob(Connection conn, Job job, String stockTableName, String stockStatsTableName)
throws SQLException, InterruptedException, IOException, ClassNotFoundException {
assertEquals("Failed to reset getRegionBoundaries counter for scanGrouper", 0,
TestingMapReduceParallelScanGrouper.getNumCallsToGetRegionBoundaries());
upsertData(conn, stockTableName);

// only run locally, rather than having to spin up a MiniMapReduce cluster and lets us use breakpoints
job.getConfiguration().set("mapreduce.framework.name", "local");

setOutput(job, stockStatsTableName);

job.setMapperClass(MapReduceIT.StockMapper.class);
job.setReducerClass(MapReduceIT.StockReducer.class);
job.setOutputFormatClass(PhoenixOutputFormat.class);

job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(DoubleWritable.class);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(MapReduceIT.StockWritable.class);

// run job and assert if success
assertTrue("Job didn't complete successfully! Check logs for reason.", job.waitForCompletion(true));
}

/**
* Custom output setting because output upsert statement setting is broken (PHOENIX-2677)
*
* @param job to update
*/
private void setOutput(Job job, String stockStatsTableName) {
final Configuration configuration = job.getConfiguration();
PhoenixConfigurationUtil.setOutputTableName(configuration, stockStatsTableName);
configuration.set(PhoenixConfigurationUtil.UPSERT_STATEMENT, "UPSERT into " + stockStatsTableName +
" (" + STOCK_NAME + ", " + MAX_RECORDING + ") values (?,?)");
job.setOutputFormatClass(PhoenixOutputFormat.class);
}


private void configureJob(Job job, String tableName, String inputQuery, String condition, boolean shouldSplit) throws Exception {
try {
upsertAndSnapshot(tableName, shouldSplit);
Expand DownExpand Up@@ -169,6 +274,22 @@ private void configureJob(Job job, String tableName, String inputQuery, String c
}
}

private void upsertData(Connection conn, String stockTableName) throws SQLException {
PreparedStatement stmt = conn.prepareStatement(String.format(UPSERT, stockTableName));
upsertData(stmt, "AAPL", 2009, new Double[]{85.88, 91.04, 88.5, 90.3});
upsertData(stmt, "AAPL", 2008, new Double[]{75.88, 81.04, 78.5, 80.3});
conn.commit();
}

private void upsertData(PreparedStatement stmt, String name, int year, Double[] data) throws SQLException {
int i = 1;
stmt.setString(i++, name);
stmt.setInt(i++, year);
Array recordings = new PhoenixArray.PrimitiveDoublePhoenixArray(PDouble.INSTANCE, data);
stmt.setArray(i++, recordings);
stmt.execute();
}

private void upsertData(String tableName) throws SQLException {
Connection conn = DriverManager.getConnection(getUrl());
PreparedStatement stmt = conn.prepareStatement(String.format(UPSERT, tableName));
Expand All@@ -192,26 +313,27 @@ private void upsertAndSnapshot(String tableName, boolean shouldSplit) throws Exc
upsertData(tableName);

TableName hbaseTableName = TableName.valueOf(tableName);
Connection conn = DriverManager.getConnection(getUrl());
HBaseAdmin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin();
try (Connection conn = DriverManager.getConnection(getUrl())) {
HBaseAdmin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin();

if (shouldSplit) {
splitTableSync(admin, hbaseTableName, "BBBB".getBytes(), 2);
}
if (shouldSplit) {
splitTableSync(admin, hbaseTableName, "BBBB".getBytes(), 2);
}

admin.snapshot(SNAPSHOT_NAME, hbaseTableName);
admin.snapshot(SNAPSHOT_NAME, hbaseTableName);

List<HBaseProtos.SnapshotDescription> snapshots = admin.listSnapshots();
Assert.assertEquals(tableName, snapshots.get(0).getTable());
List<HBaseProtos.SnapshotDescription> snapshots = admin.listSnapshots();
Assert.assertEquals(tableName, snapshots.get(0).getTable());

// Capture the snapshot timestamp to use as SCN while reading the table later
// Assigning the timestamp value here will make tests less flaky
timestamp = System.currentTimeMillis();
// Capture the snapshot timestamp to use as SCN while reading the table later
// Assigning the timestamp value here will make tests less flaky
timestamp = System.currentTimeMillis();

// upsert data after snapshot
PreparedStatement stmt = conn.prepareStatement(String.format(UPSERT, tableName));
upsertData(stmt, "DDDD", "SNFB", 45);
conn.commit();
// upsert data after snapshot
PreparedStatement stmt = conn.prepareStatement(String.format(UPSERT, tableName));
upsertData(stmt, "DDDD", "SNFB", 45);
conn.commit();
}
}

private void splitTableSync(HBaseAdmin admin, TableName hbaseTableName,
Expand Down
7 changes: 5 additions & 2 deletions phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixInputFormat.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -210,9 +210,12 @@ protected QueryPlan getQueryPlan(final JobContext context, final Configuration

// setting the snapshot configuration
String snapshotName = configuration.get(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY);
Configuration config = queryPlan.getContext().getConnection().getQueryServices().getConfiguration();
if (snapshotName != null) {
PhoenixConfigurationUtil.setSnapshotNameKey(queryPlan.getContext().getConnection().
getQueryServices().getConfiguration(), snapshotName);
PhoenixConfigurationUtil.setSnapshotNameKey(config, snapshotName);
} else {
// making sure we unset snapshot name as new job doesn't need it
config.unset(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY);
}

return queryPlan;
Expand Down