EdColeman commented on code in PR #3954:
URL: https://github.com/apache/accumulo/pull/3954#discussion_r1397799977
##########
server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader11to12.java:
##########
@@ -84,111 +89,97 @@ public void upgradeZookeeper(@NonNull ServerContext
context) {
public void upgradeRoot(@NonNull ServerContext context) {
log.debug("Upgrade root: upgrading to data version {}",
METADATA_FILE_JSON_ENCODING);
var rootName = Ample.DataLevel.METADATA.metaTable();
- processReferences(context, rootName);
+ // not using ample to avoid StoredTabletFile because old file ref is
incompatible
+ try (AccumuloClient c =
Accumulo.newClient().from(context.getProperties()).build();
+ BatchWriter batchWriter = c.createBatchWriter(rootName); Scanner
scanner =
+ new IsolatedScanner(context.createScanner(rootName,
Authorizations.EMPTY))) {
+ processReferences(batchWriter, scanner, rootName);
+ } catch (TableNotFoundException ex) {
+ throw new IllegalStateException("Failed to find table " + rootName, ex);
+ } catch (MutationsRejectedException mex) {
+ log.warn("Failed to update reference for table: " + rootName);
+ log.warn("Constraint violations: {}",
mex.getConstraintViolationSummaries());
+ throw new IllegalStateException("Failed to process table: " + rootName,
mex);
+ }
}
@Override
public void upgradeMetadata(@NonNull ServerContext context) {
log.debug("Upgrade metadata: upgrading to data version {}",
METADATA_FILE_JSON_ENCODING);
var metaName = Ample.DataLevel.USER.metaTable();
- processReferences(context, metaName);
+ try (AccumuloClient c =
Accumulo.newClient().from(context.getProperties()).build();
+ BatchWriter batchWriter = c.createBatchWriter(metaName); Scanner
scanner =
+ new IsolatedScanner(context.createScanner(metaName,
Authorizations.EMPTY))) {
+ processReferences(batchWriter, scanner, metaName);
+ } catch (TableNotFoundException ex) {
+ throw new IllegalStateException("Failed to find table " + metaName, ex);
+ } catch (MutationsRejectedException mex) {
+ log.warn("Failed to update reference for table: " + metaName);
+ log.warn("Constraint violations: {}",
mex.getConstraintViolationSummaries());
+ throw new IllegalStateException("Failed to process table: " + metaName,
mex);
+ }
}
- private void processReferences(ServerContext context, String tableName) {
- // not using ample to avoid StoredTabletFile because old file ref is
incompatible
- try (AccumuloClient c =
Accumulo.newClient().from(context.getProperties()).build();
- BatchWriter batchWriter = c.createBatchWriter(tableName); Scanner
scanner =
- new IsolatedScanner(context.createScanner(tableName,
Authorizations.EMPTY))) {
-
- scanner.fetchColumnFamily(DataFileColumnFamily.NAME);
- scanner.fetchColumnFamily(ChoppedColumnFamily.NAME);
- scanner.fetchColumnFamily(ExternalCompactionColumnFamily.NAME);
- scanner.forEach((k, v) -> {
- var family = k.getColumnFamily();
+ void processReferences(BatchWriter batchWriter, Scanner scanner, String
tableName) {
+ scanner.fetchColumnFamily(DataFileColumnFamily.NAME);
+ scanner.fetchColumnFamily(ChoppedColumnFamily.NAME);
+ scanner.fetchColumnFamily(ExternalCompactionColumnFamily.NAME);
+ try {
+ Mutation update = null;
+ for (Map.Entry<Key,Value> entry : scanner) {
+ Key key = entry.getKey();
+ Value value = entry.getValue();
+
+ // on new row, write current mutation and prepare a new one.
+ Text r = key.getRow();
+ if (update == null) {
+ update = new Mutation(r);
+ } else if (!Arrays.equals(update.getRow(), r.getBytes())) {
+ log.trace("table: {}, update: {}", tableName, update.prettyPrint());
+ batchWriter.addMutation(update);
Review Comment:
Also added test to validate batch write not called when the mutation is
empty,
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
