Metadata table bulk file columns deleted on upgrade (#4725)

Deletes the bulk file columns from the metadata table on upgrade to 4.x, if they exist.
closes #4637, related to #4587

core/src/main/java/org/apache/accumulo/core/metadata/schema/MetadataSchema.java

@@ -347,7 +347,6 @@ public static FateId getBulkLoadTid(Value v) {
       }
 
       public static FateId getBulkLoadTid(String vs) {
-        // ELASTICITY_TODO issue 4044 - May need to introduce code in upgrade to handle old format.
         return FateId.from(vs);
       }
     }

server/manager/src/main/java/org/apache/accumulo/manager/upgrade/Upgrader12to13.java

@@ -105,6 +105,8 @@ public void upgradeMetadata(ServerContext context) {
     removeMetaDataBulkLoadFilter(context, AccumuloTable.METADATA.tableId());
     LOG.info("Removing compact columns from user tables");
     removeCompactColumnsFromTable(context, AccumuloTable.METADATA.tableName());
+    LOG.info("Removing bulk file columns from metadata table");
+    removeBulkFileColumnsFromTable(context, AccumuloTable.METADATA.tableName());
   }
 
   private static void addCompactionsNode(ServerContext context) {
@@ -196,6 +198,31 @@ private void removeCompactColumnsFromTable(ServerContext context, String tableNa
     }
   }
 
+  private void removeBulkFileColumnsFromTable(ServerContext context, String tableName) {
+    // FATE transaction ids have changed from 3.x to 4.x which are used as the value for the bulk
+    // file column. FATE ops won't persist through upgrade, so these columns can be safely deleted
+    // if they exist.
+    try (var scanner = context.createScanner(tableName);
+        var writer = context.createBatchWriter(tableName)) {
+      scanner.setRange(MetadataSchema.TabletsSection.getRange());
+      scanner.fetchColumnFamily(TabletsSection.BulkFileColumnFamily.NAME);
+      for (Map.Entry<Key,Value> entry : scanner) {
+        var key = entry.getKey();
+        Mutation m = new Mutation(key.getRow());
+        Preconditions.checkState(
+            key.getColumnFamily().equals(TabletsSection.BulkFileColumnFamily.NAME),
+            "Expected family %s, saw %s ", TabletsSection.BulkFileColumnFamily.NAME,
+            key.getColumnFamily());
+        Preconditions.checkState(key.getColumnVisibilityData().length() == 0,
+            "Expected empty visibility, saw %s ", key.getColumnVisibilityData());
+        m.putDelete(key.getColumnFamily(), key.getColumnQualifier());
+        writer.addMutation(m);
+      }
+    } catch (Exception e) {
+      throw new IllegalStateException(e);
+    }
+  }
+
   private void removeUnusedZKNodes(ServerContext context) {
     try {
       final String zkRoot = ZooUtil.getRoot(context.getInstanceID());
@@ -281,7 +308,7 @@ private void deleteExternalCompactions(ServerContext context) {
         Mutation m = new Mutation(key.getRow());
         Preconditions.checkState(key.getColumnFamily().equals(ExternalCompactionColumnFamily.NAME),
             "Expected family %s, saw %s ", ExternalCompactionColumnFamily.NAME,
-            key.getColumnVisibilityData());
+            key.getColumnFamily());
         Preconditions.checkState(key.getColumnVisibilityData().length() == 0,
             "Expected empty visibility, saw %s ", key.getColumnVisibilityData());
         m.putDelete(key.getColumnFamily(), key.getColumnQualifier());