Skip to content

PHOENIX-6386 Bulkload generates unverified index rows#1144

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
stoty wants to merge 2 commits into from
Closed

PHOENIX-6386 Bulkload generates unverified index rows #1144

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
42607b0
PHOENIX-6386 Bulkload generates unverified index rows
stoty Feb 18, 2021
659279e
address review comments, and some checkstyle issues
stoty Feb 22, 2021
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
41 changes: 39 additions & 2 deletions phoenix-core/src/it/java/org/apache/phoenix/end2end/CsvBulkLoadToolIT.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -21,31 +21,43 @@
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.io.PrintWriter;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;

import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.CompareOperator;
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.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapred.FileAlreadyExistsException;
import org.apache.phoenix.hbase.index.IndexRegionObserver;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.mapreduce.CsvBulkLoadTool;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
import org.apache.phoenix.util.DateUtil;
import org.apache.phoenix.util.EncodedColumnsUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.TestUtil;
import org.junit.BeforeClass;
import org.junit.Test;
Expand DownExpand Up@@ -318,6 +330,27 @@ public void testImportWithIndex() throws Exception {

rs.close();
stmt.close();

checkIndexTableIsVerified("TABLE3_IDX");
}

private void checkIndexTableIsVerified(String indexTableName) throws SQLException, IOException {
ConnectionQueryServices cqs = conn.unwrap(PhoenixConnection.class).getQueryServices();
Table hTable = cqs.getTable(Bytes.toBytes(indexTableName));
PTable pTable = PhoenixRuntime.getTable(conn, indexTableName);

byte[] emptyKeyValueCF = SchemaUtil.getEmptyColumnFamily(pTable);
byte[] emptyKeyValueQualifier = EncodedColumnsUtil.getEmptyKeyValueInfo(pTable).getFirst();

Scan scan = new Scan();
scan.setFilter(new SingleColumnValueFilter(
emptyKeyValueCF,
emptyKeyValueQualifier,
CompareOperator.NOT_EQUAL,
new org.apache.hadoop.hbase.filter.BinaryComparator(IndexRegionObserver.VERIFIED_BYTES)));
try (ResultScanner scanner = hTable.getScanner(scan)) {
assertNull("There are non VERIFIED rows in index", scanner.next());
}
}

@Test
Expand DownExpand Up@@ -405,6 +438,10 @@ public void testImportOneIndexTable(String tableName, boolean localIndex) throws

rs.close();
stmt.close();

if (!localIndex) {
checkIndexTableIsVerified(indexTableName);
}
}

@Test
Expand Down
61 changes: 54 additions & 7 deletions phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -33,6 +33,7 @@

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
Expand All@@ -55,6 +56,7 @@
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.UpsertExecutor;
import org.apache.phoenix.hbase.index.IndexRegionObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand DownExpand Up@@ -105,6 +107,7 @@ public interface LineParser<T> {
protected PhoenixConnection conn;
protected UpsertExecutor<RECORD, ?> upsertExecutor;
protected ImportPreUpsertKeyValueProcessor preUpdateProcessor;
protected IndexStatusUpdater[] indexStatusUpdaters;
protected List<String> tableNames;
protected List<String> logicalNames;
protected MapperUpsertListener<RECORD> upsertListener;
Expand DownExpand Up@@ -176,18 +179,19 @@ record = getLineParser().parse(value.toString());
while (uncommittedDataIterator.hasNext()) {
Pair<byte[], List<Cell>> kvPair = uncommittedDataIterator.next();
List<Cell> keyValueList = kvPair.getSecond();
keyValueList = preUpdateProcessor.preUpsert(kvPair.getFirst(), keyValueList);
byte[] first = kvPair.getFirst();
byte[] tableName = kvPair.getFirst();
keyValueList = preUpdateProcessor.preUpsert(tableName, keyValueList);
// Create a list of KV for each table
for (int i = 0; i < tableNames.size(); i++) {
if (Bytes.compareTo(Bytes.toBytes(tableNames.get(i)), first) == 0) {
if (Bytes.compareTo(Bytes.toBytes(tableNames.get(i)), tableName) == 0) {
if (!map.containsKey(i)) {
map.put(i, new ArrayList<Cell>());
}
List<Cell> list = map.get(i);
for (Cell kv : keyValueList) {
list.add(kv);
List<Cell> cellsForTable = map.get(i);
if (indexStatusUpdaters[i] != null) {
indexStatusUpdaters[i].setVerified(keyValueList);
}
cellsForTable.addAll(keyValueList);
break;
}
}
Expand All@@ -210,6 +214,7 @@ record = getLineParser().parse(value.toString());
*/
private void initColumnIndexes() throws SQLException {
columnIndexes = new TreeMap<>(Bytes.BYTES_COMPARATOR);
indexStatusUpdaters = new IndexStatusUpdater[logicalNames.size()];
int columnIndex = 0;
for (int index = 0; index < logicalNames.size(); index++) {
PTable table = PhoenixRuntime.getTable(conn, logicalNames.get(index));
Expand DownExpand Up@@ -246,6 +251,10 @@ private void initColumnIndexes() throws SQLException {
byte[] cfn = Bytes.add(emptyColumnFamily, QueryConstants.NAMESPACE_SEPARATOR_BYTES, emptyKeyValue);
columnIndexes.put(cfn, new Integer(columnIndex));
columnIndex++;
if (PTable.IndexType.GLOBAL == table.getIndexType()) {
indexStatusUpdaters[index] =
new IndexStatusUpdater(emptyColumnFamily, emptyKeyValue);
}
}
}

Expand DownExpand Up@@ -411,8 +420,46 @@ public static class DefaultImportPreUpsertKeyValueProcessor implements
ImportPreUpsertKeyValueProcessor {

@Override
public List<Cell> preUpsert(byte[] rowKey, List<Cell> keyValues) {
public List<Cell> preUpsert(byte[] tableName, List<Cell> keyValues) {
return keyValues;
}
}

/**
* Updates the EMPTY cell value to VERIFIED for global index table rows.
*/
private static class IndexStatusUpdater {

private final byte[] emptyKeyValueCF;
private final int emptyKeyValueCFLength;
private final byte[] emptyKeyValueQualifier;
private final int emptyKeyValueQualifierLength;

public IndexStatusUpdater(final byte[] emptyKeyValueCF, final byte[] emptyKeyValueQualifier) {
this.emptyKeyValueCF = emptyKeyValueCF;
this.emptyKeyValueQualifier = emptyKeyValueQualifier;
emptyKeyValueCFLength = emptyKeyValueCF.length;
emptyKeyValueQualifierLength = emptyKeyValueQualifier.length;
}

/**
* Update the Empty cell values to VERIFIED in the passed keyValues list
*
* @param keyValues will be modified
*/
public void setVerified(List<Cell> keyValues) {
for (int i = 0; i < keyValues.size(); i++) {
Cell kv = keyValues.get(i);
if (CellUtil.compareFamilies(kv, emptyKeyValueCF, 0, emptyKeyValueCFLength) == 0
&& CellUtil.compareQualifiers(kv, emptyKeyValueQualifier, 0, emptyKeyValueQualifierLength) == 0) {
if (kv.getValueLength() != 1) {
//This should never happen. Fail fast if it does.
throw new IllegalArgumentException("Empty cell value length is not 1");
}
//We are directly overwriting the value for performance
kv.getValueArray()[kv.getValueOffset()] = IndexRegionObserver.VERIFIED_BYTES[0];
}
}
}
}
}
4 changes: 2 additions & 2 deletions ...nix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -39,10 +39,10 @@ public interface ImportPreUpsertKeyValueProcessor {
* Implementors can filter certain KeyValues from the list, augment the list, or return the
* same list.
*
* @param rowKey the row key for the key values that are being passed in
* @param tableName the table name for the key values that are being passed in
* @param keyValues list of KeyValues that are to be written to an HFile
* @return the list that will actually be written
*/
List<Cell> preUpsert(byte[] rowKey, List<Cell> keyValues);
List<Cell> preUpsert(byte[] tableName, List<Cell> keyValues);

}