From 42607b0c18f0120c493ea0c7fb354d511444a6cf Mon Sep 17 00:00:00 2001 From: Istvan Toth Date: Thu, 18 Feb 2021 11:36:31 +0100 Subject: [PATCH 1/2] PHOENIX-6386 Bulkload generates unverified index rows --- .../phoenix/end2end/CsvBulkLoadToolIT.java | 42 +++++++++++++ .../FormatToBytesWritableMapper.java | 60 +++++++++++++++++-- .../ImportPreUpsertKeyValueProcessor.java | 4 +- 3 files changed, 98 insertions(+), 8 deletions(-) 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..bb52b219d5a 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,13 +21,16 @@ 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; @@ -36,16 +39,30 @@ import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.CompareOperator; import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.client.Admin; +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.filter.CompareFilter; +import org.apache.hadoop.hbase.filter.SingleColumnValueFilter; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hdfs.server.namenode.SaveNamespaceCancelledException; 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.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 +335,27 @@ public void testImportWithIndex() throws Exception { rs.close(); stmt.close(); + + checkIndexTableIsVerfied("TABLE3_IDX"); + } + + private void checkIndexTableIsVerfied(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 +443,10 @@ public void testImportOneIndexTable(String tableName, boolean localIndex) throws rs.close(); stmt.close(); + + if (!localIndex) { + checkIndexTableIsVerfied(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..f493508eed6 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,17 +179,20 @@ 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); + List cellsForTable = map.get(i); + if(indexStatusUpdaters[i] != null) { + indexStatusUpdaters[i].setVerfied(keyValueList); + } for (Cell kv : keyValueList) { - list.add(kv); + cellsForTable.add(kv); } break; } @@ -210,6 +216,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 +253,9 @@ 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 +421,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 byte[] emptyKeyValueCF; + private int emptyKeyValueCFLength; + private byte[] emptyKeyValueQualifier; + private int emptyKeyValueQualifierLength; + + public IndexStatusUpdater(byte[] emptyKeyValueCF, 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 setVerfied(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); } From 659279e6cef1c3fd350f27bdfffd77fd567f40aa Mon Sep 17 00:00:00 2001 From: Istvan Toth Date: Mon, 22 Feb 2021 06:55:41 +0100 Subject: [PATCH 2/2] address review comments, and some checkstyle issues --- .../phoenix/end2end/CsvBulkLoadToolIT.java | 13 +++------ .../FormatToBytesWritableMapper.java | 29 +++++++++---------- 2 files changed, 18 insertions(+), 24 deletions(-) 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 bb52b219d5a..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 @@ -34,22 +34,16 @@ 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.CompareOperator; -import org.apache.hadoop.hbase.HConstants; -import org.apache.hadoop.hbase.client.Admin; -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.filter.CompareFilter; import org.apache.hadoop.hbase.filter.SingleColumnValueFilter; import org.apache.hadoop.hbase.util.Bytes; -import org.apache.hadoop.hdfs.server.namenode.SaveNamespaceCancelledException; import org.apache.hadoop.mapred.FileAlreadyExistsException; import org.apache.phoenix.hbase.index.IndexRegionObserver; import org.apache.phoenix.jdbc.PhoenixConnection; @@ -58,6 +52,7 @@ 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; @@ -336,10 +331,10 @@ public void testImportWithIndex() throws Exception { rs.close(); stmt.close(); - checkIndexTableIsVerfied("TABLE3_IDX"); + checkIndexTableIsVerified("TABLE3_IDX"); } - private void checkIndexTableIsVerfied(String indexTableName) throws SQLException, IOException { + 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); @@ -445,7 +440,7 @@ public void testImportOneIndexTable(String tableName, boolean localIndex) throws stmt.close(); if (!localIndex) { - checkIndexTableIsVerfied(indexTableName); + checkIndexTableIsVerified(indexTableName); } } 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 f493508eed6..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 @@ -188,12 +188,10 @@ record = getLineParser().parse(value.toString()); map.put(i, new ArrayList()); } List cellsForTable = map.get(i); - if(indexStatusUpdaters[i] != null) { - indexStatusUpdaters[i].setVerfied(keyValueList); - } - for (Cell kv : keyValueList) { - cellsForTable.add(kv); + if (indexStatusUpdaters[i] != null) { + indexStatusUpdaters[i].setVerified(keyValueList); } + cellsForTable.addAll(keyValueList); break; } } @@ -253,8 +251,9 @@ 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); + if (PTable.IndexType.GLOBAL == table.getIndexType()) { + indexStatusUpdaters[index] = + new IndexStatusUpdater(emptyColumnFamily, emptyKeyValue); } } } @@ -427,16 +426,16 @@ public List preUpsert(byte[] tableName, List keyValues) { } /** - * Updates the Empty cell value to VERIFIED for global index table rows + * Updates the EMPTY cell value to VERIFIED for global index table rows. */ private static class IndexStatusUpdater { - private byte[] emptyKeyValueCF; - private int emptyKeyValueCFLength; - private byte[] emptyKeyValueQualifier; - private int emptyKeyValueQualifierLength; + private final byte[] emptyKeyValueCF; + private final int emptyKeyValueCFLength; + private final byte[] emptyKeyValueQualifier; + private final int emptyKeyValueQualifierLength; - public IndexStatusUpdater(byte[] emptyKeyValueCF, byte[] emptyKeyValueQualifier) { + public IndexStatusUpdater(final byte[] emptyKeyValueCF, final byte[] emptyKeyValueQualifier) { this.emptyKeyValueCF = emptyKeyValueCF; this.emptyKeyValueQualifier = emptyKeyValueQualifier; emptyKeyValueCFLength = emptyKeyValueCF.length; @@ -448,8 +447,8 @@ public IndexStatusUpdater(byte[] emptyKeyValueCF, byte[] emptyKeyValueQualifier) * * @param keyValues will be modified */ - public void setVerfied(List keyValues) { - for(int i=0; i < keyValues.size() ; i++) { + 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) {