Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-15806 revert due to discussion on security

This commit is contained in:
tedyu 2016-05-24 02:08:11 -07:00
parent 2515b0974d
commit 39dc19236e
6 changed files with 168 additions and 2621 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
hbase-protocol/pom.xml
View File

@ -180,7 +180,6 @@
<include>Comparator.proto</include>
<include>Encryption.proto</include>
<include>ErrorHandling.proto</include>
<include>Export.proto</include>
<include>FS.proto</include>
<include>Filter.proto</include>
<include>HBase.proto</include>

1990
hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/ExportProtos.java
View File

File diff suppressed because it is too large Load Diff

44
hbase-protocol/src/main/protobuf/Export.proto
View File

@ -1,44 +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 hbase.pb;
option java_package = "org.apache.hadoop.hbase.protobuf.generated";
option java_outer_classname = "ExportProtos";
option java_generate_equals_and_hash = true;
option optimize_for = SPEED;
option java_generic_services = true;
import "Client.proto";
service ExportService {
rpc export (ExportRequest) returns (ExportResponse);
}
message ExportRequest {
required Scan scan = 1;
required string outputPath = 2;
optional bool compressed = 3 [default = false];
optional string compressType = 4;
optional string compressCodec = 5;
}
message ExportResponse {
required uint64 rowCount = 1;
required uint64 cellCount = 2;
}

280
hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/ExportEndpoint.java
View File

@ -1,280 +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.hadoop.hbase.coprocessor;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.classification.InterfaceStability;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
import org.apache.hadoop.hbase.ipc.ServerRpcController;
import org.apache.hadoop.hbase.mapreduce.ResultSerialization;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.protobuf.generated.ExportProtos;
import org.apache.hadoop.hbase.regionserver.Region;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.hbase.mapreduce.Export;
import org.apache.hadoop.hbase.mapreduce.Import;
/**
* Export an HBase table.
* Writes content to sequence files up in HDFS. Use {@link Import} to read it
* back in again.
* It is implemented by the endpoint technique.
* @see Export
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public class ExportEndpoint extends ExportProtos.ExportService
implements Coprocessor, CoprocessorService {
private static final Log LOG = LogFactory.getLog(ExportEndpoint.class);
private RegionCoprocessorEnvironment env = null;
@Override
public void start(CoprocessorEnvironment environment) throws IOException {
if (environment instanceof RegionCoprocessorEnvironment) {
this.env = (RegionCoprocessorEnvironment) environment;
} else {
throw new CoprocessorException("Must be loaded on a table region!");
}
}
@Override
public void stop(CoprocessorEnvironment env) throws IOException {
}
@Override
public Service getService() {
return this;
}
private static boolean getCompression(final ExportProtos.ExportRequest request) {
if (request.hasCompressed()) {
return request.getCompressed();
} else {
return false;
}
}
private static SequenceFile.CompressionType getCompressionType(final ExportProtos.ExportRequest request) {
if (!request.hasCompressType()) {
return null;
}
return SequenceFile.CompressionType.valueOf(request.getCompressType());
}
private static CompressionCodec getCompressionCodec(final Configuration conf, final ExportProtos.ExportRequest request) {
if (!request.hasCompressCodec()) {
return null;
}
try {
Class<? extends CompressionCodec> codecClass = conf.getClassByName(request.getCompressCodec()).asSubclass(CompressionCodec.class);
return ReflectionUtils.newInstance(codecClass, conf);
} catch (ClassNotFoundException e) {
throw new IllegalArgumentException("Compression codec "
+ request.getCompressCodec()+ " was not found.", e);
}
}
private static SequenceFile.Writer.Option getOutputPath(final Configuration conf,
final HRegionInfo info, final ExportProtos.ExportRequest request) throws IOException {
Path file = new Path(request.getOutputPath(), "export-" + info.getEncodedName());
FileSystem fs = FileSystem.get(conf);
if (fs.exists(file)) {
throw new IOException(file + " exists");
}
return SequenceFile.Writer.file(file);
}
private static List<SequenceFile.Writer.Option> getWriterOptions(final Configuration conf,
final HRegionInfo info, final ExportProtos.ExportRequest request) throws IOException {
List<SequenceFile.Writer.Option> rval = new LinkedList<>();
rval.add(SequenceFile.Writer.keyClass(ImmutableBytesWritable.class));
rval.add(SequenceFile.Writer.valueClass(Result.class));
rval.add(getOutputPath(conf, info, request));
boolean compressed = getCompression(request);
if (compressed) {
SequenceFile.CompressionType type = getCompressionType(request);
if (type != null) {
CompressionCodec codec = getCompressionCodec(conf, request);
rval.add(SequenceFile.Writer.compression(type, codec));
}
}
return rval;
}
private Scan validateKey(final HRegionInfo region, final Scan scan) {
byte[] regionStartKey = region.getStartKey();
byte[] originStartKey = scan.getStartRow();
if (originStartKey == null
|| Bytes.compareTo(originStartKey, regionStartKey) < 0) {
scan.setStartRow(regionStartKey);
}
byte[] regionEndKey = region.getEndKey();
byte[] originEndKey = scan.getStopRow();
if (originEndKey == null
|| Bytes.compareTo(originEndKey, regionEndKey) > 0) {
scan.setStartRow(regionEndKey);
}
return scan;
}
@Override
public void export(RpcController controller, ExportProtos.ExportRequest request,
RpcCallback<ExportProtos.ExportResponse> done) {
Region region = env.getRegion();
Configuration conf = HBaseConfiguration.create(env.getConfiguration());
conf.setStrings("io.serializations", conf.get("io.serializations"), ResultSerialization.class.getName());
try {
Scan scan = validateKey(region.getRegionInfo(), ProtobufUtil.toScan(request.getScan()));
ExportProtos.ExportResponse response = processData(conf, region, scan,
getWriterOptions(conf, region.getRegionInfo(), request));
done.run(response);
} catch (IOException e) {
ResponseConverter.setControllerException(controller, e);
LOG.error(e);
}
}
private static ExportProtos.ExportResponse processData(final Configuration conf,
final Region region, final Scan scan, final List<SequenceFile.Writer.Option> opts) throws IOException {
region.startRegionOperation();
try (SequenceFile.Writer out = SequenceFile.createWriter(conf,
opts.toArray(new SequenceFile.Writer.Option[opts.size()]));
RegionScanner scanner = region.getScanner(scan)) {
ImmutableBytesWritable key = new ImmutableBytesWritable();
long rowCount = 0;
long cellCount = 0;
List<Cell> buf = new ArrayList<>();
boolean hasMore;
do {
hasMore = scanner.nextRaw(buf);
if (!buf.isEmpty()) {
Cell firstCell = buf.get(0);
for (Cell cell : buf) {
if (Bytes.compareTo(
firstCell.getRowArray(), firstCell.getRowOffset(), firstCell.getRowLength(),
cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()) != 0) {
throw new IOException("Why the RegionScanner#nextRaw returns the data of different rows??");
}
}
key.set(firstCell.getRowArray(), firstCell.getRowOffset(), firstCell.getRowLength());
out.append(key, Result.create(buf));
++rowCount;
cellCount += buf.size();
buf.clear();
}
} while (hasMore);
return ExportProtos.ExportResponse.newBuilder()
.setRowCount(rowCount)
.setCellCount(cellCount)
.build();
} finally {
region.closeRegionOperation();
}
}
public static void main(String[] args) throws IOException, Throwable {
run(HBaseConfiguration.create(), args);
}
public static Map<byte[], ExportProtos.ExportResponse> run(final Configuration conf,
final String[] args) throws IOException, Throwable {
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (!Export.checkArguments(otherArgs)) {
Export.usage("Wrong number of arguments: " + otherArgs.length);
System.exit(-1);
}
TableName tableName = TableName.valueOf(otherArgs[0]);
FileSystem fs = FileSystem.get(conf);
String dir = otherArgs[1];
checkDir(fs, dir);
Scan scan = Export.getConfiguredScanForJob(conf, otherArgs);
final ExportProtos.ExportRequest request = getConfiguredRequestForJob(conf, otherArgs, scan);
try (Connection con = ConnectionFactory.createConnection(conf);
Table table = con.getTable(tableName)) {
return table.coprocessorService(ExportProtos.ExportService.class,
scan.getStartRow(),
scan.getStopRow(), new Batch.Call<ExportProtos.ExportService, ExportProtos.ExportResponse>() {
@Override
public ExportProtos.ExportResponse call(ExportProtos.ExportService service) throws IOException {
ServerRpcController controller = new ServerRpcController();
BlockingRpcCallback<ExportProtos.ExportResponse> rpcCallback = new BlockingRpcCallback<>();
service.export(controller, request, rpcCallback);
if (controller.failedOnException()) {
throw controller.getFailedOn();
}
return rpcCallback.get();
}
});
} catch (Throwable e) {
fs.delete(new Path(dir), true);
throw e;
}
}
private static void checkDir(final FileSystem fs, final String path) throws IOException {
Path dir = fs.makeQualified(new Path(path));
if (fs.exists(dir)) {
throw new RuntimeException("The " + path + " exists");
}
fs.mkdirs(dir);
fs.setPermission(dir, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
}
private static ExportProtos.ExportRequest getConfiguredRequestForJob(Configuration conf,
String[] args, final Scan scan) throws IOException {
String dir = args[1];
boolean compressed = conf.getBoolean(FileOutputFormat.COMPRESS, true);
String compressionType = conf.get(FileOutputFormat.COMPRESS_TYPE,
SequenceFile.CompressionType.RECORD.toString());
String compressionCodec = conf.get(FileOutputFormat.COMPRESS_CODEC,
DefaultCodec.class.getName());
LOG.info("compressed=" + compressed
+ ", compression type=" + compressionType
+ ", compression codec=" + compressionCodec);
return ExportProtos.ExportRequest.newBuilder()
.setScan(ProtobufUtil.toScan(scan))
.setOutputPath(dir)
.setCompressed(compressed)
.setCompressCodec(compressionCodec)
.setCompressType(compressionType)
.build();
}
}

11
hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/Export.java
View File

@ -86,7 +86,7 @@ public class Export extends Configured implements Tool {
return job;
}
public static Scan getConfiguredScanForJob(Configuration conf, String[] args) throws IOException {
private static Scan getConfiguredScanForJob(Configuration conf, String[] args) throws IOException {
Scan s = new Scan();
// Optional arguments.
// Set Scan Versions
@ -150,7 +150,7 @@ public class Export extends Configured implements Tool {
/*
* @param errorMsg Error message. Can be null.
*/
public static void usage(final String errorMsg) {
private static void usage(final String errorMsg) {
if (errorMsg != null && errorMsg.length() > 0) {
System.err.println("ERROR: " + errorMsg);
}
@ -176,12 +176,11 @@ public class Export extends Configured implements Tool {
System.err.println("For tables with very wide rows consider setting the batch size as below:\n"
+ " -D" + EXPORT_BATCHING + "=10");
}
public static boolean checkArguments(final String[] args) {
return args.length >= 2;
}
@Override
public int run(String[] args) throws Exception {
if (!checkArguments(args)) {
if (args.length < 2) {
usage("Wrong number of arguments: " + args.length);
return -1;
}

463
hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestImportExport.java
View File

@ -17,7 +17,13 @@
*/
package org.apache.hadoop.hbase.mapreduce;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.ByteArrayOutputStream;
@ -28,10 +34,11 @@ import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.NavigableMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HConstants;
@ -52,8 +59,6 @@ import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
import org.apache.hadoop.hbase.coprocessor.ExportEndpoint;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterBase;
import org.apache.hadoop.hbase.filter.PrefixFilter;
@ -67,10 +72,7 @@ import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.testclassification.VerySlowMapReduceTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.LauncherSecurityManager;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.ToolRunner;
import org.junit.After;
import org.junit.AfterClass;
@ -83,12 +85,6 @@ import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
/**
* Tests the table import and table export MR job functionality
@ -110,40 +106,14 @@ public class TestImportExport {
private static final String EXPORT_BATCH_SIZE = "100";
private static long now = System.currentTimeMillis();
private static final Exporter EXPORTER_MR = new Exporter() {
@Override
public boolean runExport(String[] args) throws Throwable {
// need to make a copy of the configuration because to make sure different temp dirs are used.
int status = ToolRunner.run(new Configuration(UTIL.getConfiguration()), new Export(), args);
return status == 0;
}
@Override
public String toString() {
return "MR-based export";
}
};
private static final Exporter EXPORTER_ENDPOINT = new Exporter() {
@Override
public boolean runExport(String[] args) throws Throwable {
ExportEndpoint.run(new Configuration(UTIL.getConfiguration()), args);
return true;
}
@Override
public String toString() {
return "Endpoint-based export";
}
};
private static final List<Exporter> EXPORTERS = Arrays.asList(EXPORTER_MR, EXPORTER_ENDPOINT);
@BeforeClass
public static void beforeClass() throws Exception {
// Up the handlers; this test needs more than usual.
UTIL.getConfiguration().setStrings(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,
ExportEndpoint.class.getName());
UTIL.getConfiguration().setInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT, 10);
UTIL.startMiniCluster();
FileSystem fs = FileSystem.get(UTIL.getConfiguration());
FQ_OUTPUT_DIR =
new Path(OUTPUT_DIR).makeQualified(fs.getUri(), fs.getWorkingDirectory()).toString();
new Path(OUTPUT_DIR).makeQualified(FileSystem.get(UTIL.getConfiguration())).toString();
}
@AfterClass
@ -161,108 +131,47 @@ public class TestImportExport {
@Before
@After
public void cleanup() throws IOException {
deleteOutput();
}
private static void deleteOutput() throws IOException {
public void cleanup() throws Exception {
FileSystem fs = FileSystem.get(UTIL.getConfiguration());
fs.delete(new Path(OUTPUT_DIR), true);
}
/**
* Runs an export job with the specified command line args
* @param args
* @return true if job completed successfully
* @throws IOException
* @throws InterruptedException
* @throws ClassNotFoundException
*/
boolean runExport(String[] args) throws Exception {
// need to make a copy of the configuration because to make sure different temp dirs are used.
int status = ToolRunner.run(new Configuration(UTIL.getConfiguration()), new Export(), args);
return status == 0;
}
/**
* Runs an import job with the specified command line args
* @param args
* @return true if job completed successfully
* @throws Exception
* @throws IOException
* @throws InterruptedException
* @throws ClassNotFoundException
*/
boolean runImport(String[] args) throws Exception {
// need to make a copy of the configuration because to make sure different temp dirs are used.
int status = ToolRunner.run(new Configuration(UTIL.getConfiguration()), new Import(), args);
return status == 0;
}
/**
* Test the writer's options.
* @throws IOException
*/
@Test
public void testOutputFileFormat() throws IOException, Throwable {
String exportTable = "testOutputFileFormat";
for (Exporter exporter : EXPORTERS) {
testOutputFileFormat(exportTable, exporter);
UTIL.deleteTable(TableName.valueOf(exportTable));
deleteOutput();
}
}
/**
* Test the writer's options.
* @throws IOException
*/
public void testOutputFileFormat(final String exportTable, final Exporter exporter) throws IOException, Throwable {
String codec = BZip2Codec.class.getName();
String type = SequenceFile.CompressionType.RECORD.name();
try (Table t = UTIL.createTable(TableName.valueOf(exportTable), FAMILYA, 3);) {
Put p = new Put(ROW1);
p.addColumn(FAMILYA, QUAL, now, QUAL);
t.put(p);
p = new Put(ROW2);
p.addColumn(FAMILYA, QUAL, now, QUAL);
t.put(p);
p = new Put(ROW3);
p.addColumn(FAMILYA, QUAL, now, QUAL);
t.put(p);
}
//use compress
String[] args = new String[] {
// Only export row1 & row2.
"-D" + FileOutputFormat.COMPRESS + "=true",
"-D" + FileOutputFormat.COMPRESS_CODEC + "=" + codec,
"-D" + FileOutputFormat.COMPRESS_TYPE + "=" + type,
exportTable,
FQ_OUTPUT_DIR
};
assertTrue(exporter.toString(), exporter.runExport(args));
FileSystem fs = FileSystem.get(UTIL.getConfiguration());
List<FileStatus> files = Arrays.asList(fs.listStatus(new Path(FQ_OUTPUT_DIR)));
assertEquals(exporter.toString(), false, files.isEmpty());
Configuration copy = new Configuration(UTIL.getConfiguration());
//need to make a copy of the configuration because to make sure the Exporter has set the "io.serializations"
copy.setStrings("io.serializations", copy.get("io.serializations"),
ResultSerialization.class.getName());
for (FileStatus file : files) {
Path path = file.getPath();
//skips the MR meta output
if (path.getName().equals("_SUCCESS")) {
continue;
}
try (SequenceFile.Reader reader = new SequenceFile.Reader(
copy, SequenceFile.Reader.file(file.getPath()))) {
assertEquals(exporter.toString(), reader.getCompressionCodec().getClass().getName(), codec);
assertEquals(exporter.toString(), reader.getCompressionType().name(), type);
}
}
}
/**
* Test simple replication case with column mapping
* @throws IOException
* @throws Exception
*/
@Test
public void testSimpleCase() throws IOException, Throwable {
String exportTable = "exportSimpleCase";
String importTable = "importTableSimpleCase";
for (Exporter exporter : EXPORTERS) {
testSimpleCase(exportTable, importTable, exporter);
UTIL.deleteTable(TableName.valueOf(exportTable));
UTIL.deleteTable(TableName.valueOf(importTable));
deleteOutput();
}
}
/**
* Test simple replication case with column mapping.
*/
public void testSimpleCase(final String exportTable, final String importTable,
final Exporter exporter) throws IOException, Throwable {
try (Table t = UTIL.createTable(TableName.valueOf(exportTable), FAMILYA, 3);) {
public void testSimpleCase() throws Exception {
String EXPORT_TABLE = "exportSimpleCase";
try (Table t = UTIL.createTable(TableName.valueOf(EXPORT_TABLE), FAMILYA, 3);) {
Put p = new Put(ROW1);
p.addColumn(FAMILYA, QUAL, now, QUAL);
p.addColumn(FAMILYA, QUAL, now + 1, QUAL);
@ -280,53 +189,53 @@ public class TestImportExport {
t.put(p);
}
String[] args = new String[] {
// Only export row1 & row2.
"-D" + TableInputFormat.SCAN_ROW_START + "=\\x32row1",
"-D" + TableInputFormat.SCAN_ROW_STOP + "=\\x32row3",
exportTable,
FQ_OUTPUT_DIR,
"1000", // max number of key versions per key to export
};
assertTrue(exporter.toString(), exporter.runExport(args));
try (Table t = UTIL.createTable(TableName.valueOf(importTable), FAMILYB, 3);) {
args = new String[] {
"-D" + Import.CF_RENAME_PROP + "="+FAMILYA_STRING+":"+FAMILYB_STRING,
importTable,
FQ_OUTPUT_DIR
String[] args = new String[] {
// Only export row1 & row2.
"-D" + TableInputFormat.SCAN_ROW_START + "=\\x32row1",
"-D" + TableInputFormat.SCAN_ROW_STOP + "=\\x32row3",
EXPORT_TABLE,
FQ_OUTPUT_DIR,
"1000", // max number of key versions per key to export
};
assertTrue(exporter.toString(), runImport(args));
assertTrue(runExport(args));
Get g = new Get(ROW1);
g.setMaxVersions();
Result r = t.get(g);
assertEquals(exporter.toString(), 3, r.size());
g = new Get(ROW2);
g.setMaxVersions();
r = t.get(g);
assertEquals(exporter.toString(), 3, r.size());
g = new Get(ROW3);
r = t.get(g);
assertEquals(exporter.toString(), 0, r.size());
}
String IMPORT_TABLE = "importTableSimpleCase";
try (Table t = UTIL.createTable(TableName.valueOf(IMPORT_TABLE), FAMILYB, 3);) {
args = new String[] {
"-D" + Import.CF_RENAME_PROP + "="+FAMILYA_STRING+":"+FAMILYB_STRING,
IMPORT_TABLE,
FQ_OUTPUT_DIR
};
assertTrue(runImport(args));
Get g = new Get(ROW1);
g.setMaxVersions();
Result r = t.get(g);
assertEquals(3, r.size());
g = new Get(ROW2);
g.setMaxVersions();
r = t.get(g);
assertEquals(3, r.size());
g = new Get(ROW3);
r = t.get(g);
assertEquals(0, r.size());
}
}
/**
* Test export hbase:meta table
*
* @throws IOException
* @throws Exception
*/
@Test
public void testMetaExport() throws IOException, Throwable {
String exportTable = TableName.META_TABLE_NAME.getNameAsString();
String[] args = new String[] { exportTable, FQ_OUTPUT_DIR, "1", "0", "0" };
for (Exporter exporter : EXPORTERS) {
assertTrue(exporter.toString(), exporter.runExport(args));
deleteOutput();
}
public void testMetaExport() throws Exception {
String EXPORT_TABLE = TableName.META_TABLE_NAME.getNameAsString();
String[] args = new String[] { EXPORT_TABLE, FQ_OUTPUT_DIR, "1", "0", "0" };
assertTrue(runExport(args));
}
/**
* Test import data from 0.94 exported file.
* Test import data from 0.94 exported file
* @throws Exception
*/
@Test
@ -343,11 +252,11 @@ public class TestImportExport {
Path importPath = new Path(f.toURI());
FileSystem fs = FileSystem.get(UTIL.getConfiguration());
fs.copyFromLocalFile(importPath, new Path(FQ_OUTPUT_DIR + Path.SEPARATOR + name));
String importTable = name;
try (Table t = UTIL.createTable(TableName.valueOf(importTable), Bytes.toBytes("f1"), 3);) {
String IMPORT_TABLE = name;
try (Table t = UTIL.createTable(TableName.valueOf(IMPORT_TABLE), Bytes.toBytes("f1"), 3);) {
String[] args = new String[] {
"-Dhbase.import.version=0.94" ,
importTable, FQ_OUTPUT_DIR
IMPORT_TABLE, FQ_OUTPUT_DIR
};
assertTrue(runImport(args));
/* exportedTableIn94Format contains 5 rows
@ -361,28 +270,18 @@ public class TestImportExport {
assertEquals(5, UTIL.countRows(t));
}
}
/**
* Test export scanner batching
* @throws java.lang.IOException
*/
@Test
public void testExportScannerBatching() throws IOException, Throwable {
String exportTable = "exportWithBatch";
for (Exporter exporter : EXPORTERS) {
testExportScannerBatching(exportTable, exporter);
UTIL.deleteTable(TableName.valueOf(exportTable));
deleteOutput();
}
}
/**
* Test export scanner batching.
*/
public void testExportScannerBatching(final String exportTable, final Exporter exporter) throws IOException, Throwable {
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(exportTable));
public void testExportScannerBatching() throws Exception {
String BATCH_TABLE = "exportWithBatch";
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(BATCH_TABLE));
desc.addFamily(new HColumnDescriptor(FAMILYA)
.setMaxVersions(1)
);
UTIL.getAdmin().createTable(desc);
UTIL.getHBaseAdmin().createTable(desc);
try (Table t = UTIL.getConnection().getTable(desc.getTableName());) {
Put p = new Put(ROW1);
@ -395,34 +294,25 @@ public class TestImportExport {
String[] args = new String[] {
"-D" + Export.EXPORT_BATCHING + "=" + EXPORT_BATCH_SIZE, // added scanner batching arg.
exportTable,
BATCH_TABLE,
FQ_OUTPUT_DIR
};
assertTrue(exporter.toString(), exporter.runExport(args));
assertTrue(runExport(args));
FileSystem fs = FileSystem.get(UTIL.getConfiguration());
fs.delete(new Path(FQ_OUTPUT_DIR), true);
}
}
@Test
public void testWithDeletes() throws IOException, Throwable {
String exportTable = "exportWithDeletes";
String importTable = "importWithDeletes";
for (Exporter exporter : EXPORTERS) {
testWithDeletes(exportTable, importTable, exporter);
UTIL.deleteTable(TableName.valueOf(exportTable));
UTIL.deleteTable(TableName.valueOf(importTable));
deleteOutput();
}
}
public void testWithDeletes(final String exportTable, final String importTable,
final Exporter exporter) throws IOException, Throwable {
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(exportTable));
public void testWithDeletes() throws Exception {
String EXPORT_TABLE = "exportWithDeletes";
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(EXPORT_TABLE));
desc.addFamily(new HColumnDescriptor(FAMILYA)
.setMaxVersions(5)
.setKeepDeletedCells(KeepDeletedCells.TRUE)
);
UTIL.getAdmin().createTable(desc);
UTIL.getHBaseAdmin().createTable(desc);
try (Table t = UTIL.getConnection().getTable(desc.getTableName());) {
Put p = new Put(ROW1);
@ -442,23 +332,25 @@ public class TestImportExport {
String[] args = new String[] {
"-D" + Export.RAW_SCAN + "=true",
exportTable,
EXPORT_TABLE,
FQ_OUTPUT_DIR,
"1000", // max number of key versions per key to export
};
assertTrue(exporter.toString(), exporter.runExport(args));
desc = new HTableDescriptor(TableName.valueOf(importTable));
assertTrue(runExport(args));
String IMPORT_TABLE = "importWithDeletes";
desc = new HTableDescriptor(TableName.valueOf(IMPORT_TABLE));
desc.addFamily(new HColumnDescriptor(FAMILYA)
.setMaxVersions(5)
.setKeepDeletedCells(KeepDeletedCells.TRUE)
);
UTIL.getAdmin().createTable(desc);
UTIL.getHBaseAdmin().createTable(desc);
try (Table t = UTIL.getConnection().getTable(desc.getTableName());) {
args = new String[] {
importTable,
FQ_OUTPUT_DIR
IMPORT_TABLE,
FQ_OUTPUT_DIR
};
assertTrue(exporter.toString(), runImport(args));
assertTrue(runImport(args));
Scan s = new Scan();
s.setMaxVersions();
@ -466,36 +358,29 @@ public class TestImportExport {
ResultScanner scanner = t.getScanner(s);
Result r = scanner.next();
Cell[] res = r.rawCells();
assertTrue(exporter.toString(), CellUtil.isDeleteFamily(res[0]));
assertEquals(exporter.toString(), now+4, res[1].getTimestamp());
assertEquals(exporter.toString(), now+3, res[2].getTimestamp());
assertTrue(exporter.toString(), CellUtil.isDelete(res[3]));
assertEquals(exporter.toString(), now+2, res[4].getTimestamp());
assertEquals(exporter.toString(), now+1, res[5].getTimestamp());
assertEquals(exporter.toString(), now, res[6].getTimestamp());
assertTrue(CellUtil.isDeleteFamily(res[0]));
assertEquals(now+4, res[1].getTimestamp());
assertEquals(now+3, res[2].getTimestamp());
assertTrue(CellUtil.isDelete(res[3]));
assertEquals(now+2, res[4].getTimestamp());
assertEquals(now+1, res[5].getTimestamp());
assertEquals(now, res[6].getTimestamp());
}
}
@Test
public void testWithMultipleDeleteFamilyMarkersOfSameRowSameFamily() throws IOException, Throwable {
String exportTable = "exportWithMultipleDeleteFamilyMarkersOfSameRowSameFamily";
String importTable = "importWithMultipleDeleteFamilyMarkersOfSameRowSameFamily";
for (Exporter exporter : EXPORTERS) {
testWithMultipleDeleteFamilyMarkersOfSameRowSameFamily(exportTable, importTable, exporter);
UTIL.deleteTable(TableName.valueOf(exportTable));
UTIL.deleteTable(TableName.valueOf(importTable));
deleteOutput();
}
}
public void testWithMultipleDeleteFamilyMarkersOfSameRowSameFamily(final String exportTable, final String importTable,
final Exporter exporter) throws IOException, Throwable {
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(exportTable));
public void testWithMultipleDeleteFamilyMarkersOfSameRowSameFamily() throws Exception {
TableName EXPORT_TABLE =
TableName.valueOf("exportWithMultipleDeleteFamilyMarkersOfSameRowSameFamily");
HTableDescriptor desc = new HTableDescriptor(EXPORT_TABLE);
desc.addFamily(new HColumnDescriptor(FAMILYA)
.setMaxVersions(5)
.setKeepDeletedCells(KeepDeletedCells.TRUE)
);
UTIL.getAdmin().createTable(desc);
UTIL.getHBaseAdmin().createTable(desc);
Table exportT = UTIL.getConnection().getTable(desc.getTableName());
Table exportT = UTIL.getConnection().getTable(EXPORT_TABLE);
//Add first version of QUAL
Put p = new Put(ROW1);
@ -517,24 +402,26 @@ public class TestImportExport {
String[] args = new String[] {
"-D" + Export.RAW_SCAN + "=true", exportTable,
"-D" + Export.RAW_SCAN + "=true", EXPORT_TABLE.getNameAsString(),
FQ_OUTPUT_DIR,
"1000", // max number of key versions per key to export
};
assertTrue(exporter.toString(), exporter.runExport(args));
desc = new HTableDescriptor(TableName.valueOf(importTable));
assertTrue(runExport(args));
String IMPORT_TABLE = "importWithMultipleDeleteFamilyMarkersOfSameRowSameFamily";
desc = new HTableDescriptor(TableName.valueOf(IMPORT_TABLE));
desc.addFamily(new HColumnDescriptor(FAMILYA)
.setMaxVersions(5)
.setKeepDeletedCells(KeepDeletedCells.TRUE)
);
UTIL.getAdmin().createTable(desc);
UTIL.getHBaseAdmin().createTable(desc);
Table importT = UTIL.getConnection().getTable(TableName.valueOf(importTable));
Table importT = UTIL.getConnection().getTable(TableName.valueOf(IMPORT_TABLE));
args = new String[] {
importTable,
IMPORT_TABLE,
FQ_OUTPUT_DIR
};
assertTrue(exporter.toString(), runImport(args));
assertTrue(runImport(args));
Scan s = new Scan();
s.setMaxVersions();
@ -547,39 +434,26 @@ public class TestImportExport {
Result exportedTResult = exportedTScanner.next();
try {
Result.compareResults(exportedTResult, importedTResult);
} catch (IOException e) {
} catch (Exception e) {
fail("Original and imported tables data comparision failed with error:"+e.getMessage());
} finally {
exportT.close();
importT.close();
}
}
/**
* Create a simple table, run an Export Job on it, Import with filtering on, verify counts,
* attempt with invalid values.
*/
@Test
public void testWithFilter() throws IOException, Throwable {
String exportTable = "exportSimpleCase_ImportWithFilter";
String importTable = "importWithFilter";
for (Exporter exporter : EXPORTERS) {
testWithFilter(exportTable, importTable, exporter);
UTIL.deleteTable(TableName.valueOf(exportTable));
UTIL.deleteTable(TableName.valueOf(importTable));
deleteOutput();
}
}
/**
* Create a simple table, run an Export Job on it, Import with filtering on, verify counts,
* attempt with invalid values.
*/
public void testWithFilter(final String exportTable, final String importTable,
final Exporter exporter) throws IOException, Throwable {
public void testWithFilter() throws Exception {
// Create simple table to export
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(exportTable));
String EXPORT_TABLE = "exportSimpleCase_ImportWithFilter";
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(EXPORT_TABLE));
desc.addFamily(new HColumnDescriptor(FAMILYA).setMaxVersions(5));
UTIL.getAdmin().createTable(desc);
Table exportT = UTIL.getConnection().getTable(desc.getTableName());
UTIL.getHBaseAdmin().createTable(desc);
Table exportTable = UTIL.getConnection().getTable(desc.getTableName());
Put p1 = new Put(ROW1);
p1.addColumn(FAMILYA, QUAL, now, QUAL);
@ -592,42 +466,43 @@ public class TestImportExport {
Put p2 = new Put(ROW2);
p2.addColumn(FAMILYA, QUAL, now, QUAL);
exportT.put(Arrays.asList(p1, p2));
exportTable.put(Arrays.asList(p1, p2));
// Export the simple table
String[] args = new String[] { exportTable, FQ_OUTPUT_DIR, "1000" };
assertTrue(exporter.toString(), exporter.runExport(args));
String[] args = new String[] { EXPORT_TABLE, FQ_OUTPUT_DIR, "1000" };
assertTrue(runExport(args));
// Import to a new table
desc = new HTableDescriptor(TableName.valueOf(importTable));
String IMPORT_TABLE = "importWithFilter";
desc = new HTableDescriptor(TableName.valueOf(IMPORT_TABLE));
desc.addFamily(new HColumnDescriptor(FAMILYA).setMaxVersions(5));
UTIL.getAdmin().createTable(desc);
UTIL.getHBaseAdmin().createTable(desc);
Table importT = UTIL.getConnection().getTable(desc.getTableName());
Table importTable = UTIL.getConnection().getTable(desc.getTableName());
args = new String[] { "-D" + Import.FILTER_CLASS_CONF_KEY + "=" + PrefixFilter.class.getName(),
"-D" + Import.FILTER_ARGS_CONF_KEY + "=" + Bytes.toString(ROW1), importTable,
"-D" + Import.FILTER_ARGS_CONF_KEY + "=" + Bytes.toString(ROW1), IMPORT_TABLE,
FQ_OUTPUT_DIR,
"1000" };
assertTrue(exporter.toString(), runImport(args));
assertTrue(runImport(args));
// get the count of the source table for that time range
PrefixFilter filter = new PrefixFilter(ROW1);
int count = getCount(exportT, filter);
int count = getCount(exportTable, filter);
Assert.assertEquals("Unexpected row count between export(" + exporter.toString() + ") and import tables", count,
getCount(importT, null));
Assert.assertEquals("Unexpected row count between export and import tables", count,
getCount(importTable, null));
// and then test that a broken command doesn't bork everything - easier here because we don't
// need to re-run the export job
args = new String[] { "-D" + Import.FILTER_CLASS_CONF_KEY + "=" + Filter.class.getName(),
"-D" + Import.FILTER_ARGS_CONF_KEY + "=" + Bytes.toString(ROW1) + "", exportTable,
"-D" + Import.FILTER_ARGS_CONF_KEY + "=" + Bytes.toString(ROW1) + "", EXPORT_TABLE,
FQ_OUTPUT_DIR, "1000" };
assertFalse(runImport(args));
// cleanup
exportT.close();
importT.close();
exportTable.close();
importTable.close();
}
/**
@ -762,50 +637,40 @@ public class TestImportExport {
Import.addFilterAndArguments(configuration, FilterBase.class, args);
assertEquals("org.apache.hadoop.hbase.filter.FilterBase",
configuration.get(Import.FILTER_CLASS_CONF_KEY));
configuration.get(Import.FILTER_CLASS_CONF_KEY));
assertEquals("param1,param2", configuration.get(Import.FILTER_ARGS_CONF_KEY));
}
@Test
public void testDurability() throws IOException, Throwable {
String exportTable = "exporttestDurability";
String importTable = "importTestDurability1";
String importTableV2 = "importTestDurability2";
for (Exporter exporter : EXPORTERS) {
testDurability(exportTable, importTable, importTableV2, exporter);
UTIL.deleteTable(TableName.valueOf(exportTable));
UTIL.deleteTable(TableName.valueOf(importTable));
UTIL.deleteTable(TableName.valueOf(importTableV2));
deleteOutput();
}
}
public void testDurability(final String exportTable, final String importTable, final String importTable2,
final Exporter exporter) throws IOException, Throwable {
public void testDurability() throws Exception {
// Create an export table.
try (Table exportT = UTIL.createTable(TableName.valueOf(exportTable), FAMILYA, 3);) {
String exportTableName = "exporttestDurability";
try (Table exportTable = UTIL.createTable(TableName.valueOf(exportTableName), FAMILYA, 3);) {
// Insert some data
Put put = new Put(ROW1);
put.addColumn(FAMILYA, QUAL, now, QUAL);
put.addColumn(FAMILYA, QUAL, now + 1, QUAL);
put.addColumn(FAMILYA, QUAL, now + 2, QUAL);
exportT.put(put);
exportTable.put(put);
put = new Put(ROW2);
put.addColumn(FAMILYA, QUAL, now, QUAL);
put.addColumn(FAMILYA, QUAL, now + 1, QUAL);
put.addColumn(FAMILYA, QUAL, now + 2, QUAL);
exportT.put(put);
exportTable.put(put);
// Run the export
String[] args = new String[] { exportTable, FQ_OUTPUT_DIR, "1000"};
assertTrue(exporter.toString(), exporter.runExport(args));
String[] args = new String[] { exportTableName, FQ_OUTPUT_DIR, "1000"};
assertTrue(runExport(args));
// Create the table for import
Table importT = UTIL.createTable(TableName.valueOf(importTable), FAMILYA, 3);
String importTableName = "importTestDurability1";
Table importTable = UTIL.createTable(TableName.valueOf(importTableName), FAMILYA, 3);
// Register the wal listener for the import table
HRegionInfo region = UTIL.getHBaseCluster().getRegionServerThreads().get(0).getRegionServer()
.getOnlineRegions(importT.getName()).get(0).getRegionInfo();
.getOnlineRegions(importTable.getName()).get(0).getRegionInfo();
TableWALActionListener walListener = new TableWALActionListener(region);
WAL wal = UTIL.getMiniHBaseCluster().getRegionServer(0).getWAL(region);
wal.registerWALActionsListener(walListener);
@ -813,26 +678,27 @@ public class TestImportExport {
// Run the import with SKIP_WAL
args =
new String[] { "-D" + Import.WAL_DURABILITY + "=" + Durability.SKIP_WAL.name(),
importTable, FQ_OUTPUT_DIR };
assertTrue(exporter.toString(), runImport(args));
importTableName, FQ_OUTPUT_DIR };
assertTrue(runImport(args));
//Assert that the wal is not visisted
assertTrue(exporter.toString(), !walListener.isWALVisited());
assertTrue(!walListener.isWALVisited());
//Ensure that the count is 2 (only one version of key value is obtained)
assertTrue(exporter.toString(), getCount(importT, null) == 2);
assertTrue(getCount(importTable, null) == 2);
// Run the import with the default durability option
importT = UTIL.createTable(TableName.valueOf(importTable2), FAMILYA, 3);
importTableName = "importTestDurability2";
importTable = UTIL.createTable(TableName.valueOf(importTableName), FAMILYA, 3);
region = UTIL.getHBaseCluster().getRegionServerThreads().get(0).getRegionServer()
.getOnlineRegions(importT.getName()).get(0).getRegionInfo();
.getOnlineRegions(importTable.getName()).get(0).getRegionInfo();
wal = UTIL.getMiniHBaseCluster().getRegionServer(0).getWAL(region);
walListener = new TableWALActionListener(region);
wal.registerWALActionsListener(walListener);
args = new String[] { importTable2, FQ_OUTPUT_DIR };
assertTrue(exporter.toString(), runImport(args));
args = new String[] { importTableName, FQ_OUTPUT_DIR };
assertTrue(runImport(args));
//Assert that the wal is visisted
assertTrue(exporter.toString(), walListener.isWALVisited());
assertTrue(walListener.isWALVisited());
//Ensure that the count is 2 (only one version of key value is obtained)
assertTrue(exporter.toString(), getCount(importT, null) == 2);
assertTrue(getCount(importTable, null) == 2);
}
}
@ -861,7 +727,4 @@ public class TestImportExport {
return isVisited;
}
}
public interface Exporter {
boolean runExport(final String[] args) throws Throwable;
}
}