diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/CsvBulkLoadToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CsvBulkLoadToolIT.java index 81c22552122..5e0200bde8b 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/CsvBulkLoadToolIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CsvBulkLoadToolIT.java @@ -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; @@ -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 @@ -405,6 +438,10 @@ public void testImportOneIndexTable(String tableName, boolean localIndex) throws rs.close(); stmt.close(); + + if (!localIndex) { + checkIndexTableIsVerified(indexTableName); + } } @Test diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java index 40d0fb26e54..31ab60b9e54 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/FormatToBytesWritableMapper.java @@ -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; @@ -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; @@ -105,6 +107,7 @@ public interface LineParser { protected PhoenixConnection conn; protected UpsertExecutor upsertExecutor; protected ImportPreUpsertKeyValueProcessor preUpdateProcessor; + protected IndexStatusUpdater[] indexStatusUpdaters; protected List tableNames; protected List logicalNames; protected MapperUpsertListener upsertListener; @@ -176,18 +179,19 @@ record = getLineParser().parse(value.toString()); while (uncommittedDataIterator.hasNext()) { Pair> kvPair = uncommittedDataIterator.next(); List 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()); } - List list = map.get(i); - for (Cell kv : keyValueList) { - list.add(kv); + List cellsForTable = map.get(i); + if (indexStatusUpdaters[i] != null) { + indexStatusUpdaters[i].setVerified(keyValueList); } + cellsForTable.addAll(keyValueList); break; } } @@ -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)); @@ -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); + } } } @@ -411,8 +420,46 @@ public static class DefaultImportPreUpsertKeyValueProcessor implements ImportPreUpsertKeyValueProcessor { @Override - public List preUpsert(byte[] rowKey, List keyValues) { + public List preUpsert(byte[] tableName, List 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 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]; + } + } + } + } } diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java index dccbe7acf25..2de79342b36 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/ImportPreUpsertKeyValueProcessor.java @@ -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 preUpsert(byte[] rowKey, List keyValues); + List preUpsert(byte[] tableName, List keyValues); }