Changed encoding of input files from Base64 to a list of objects

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

@@ -18,24 +18,21 @@
  */
 package org.apache.accumulo.core.metadata.schema;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toSet;
 import static org.apache.accumulo.core.util.LazySingletons.GSON;
 
-import java.util.Base64;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
 
+import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.fate.FateId;
 import org.apache.accumulo.core.metadata.ReferencedTabletFile;
 import org.apache.accumulo.core.metadata.StoredTabletFile;
 import org.apache.accumulo.core.spi.compaction.CompactionKind;
 import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
-
-import com.google.gson.Gson;
-import com.google.gson.reflect.TypeToken;
+import org.apache.hadoop.fs.Path;
 
 public class CompactionMetadata {
 
@@ -98,10 +95,31 @@ public FateId getFateId() {
     return fateId;
   }
 
+  private static class InputFile {
+    String path;
+    String startRow;
+    String endRow;
+
+    private static InputFile from(StoredTabletFile stf) {
+      InputFile i = new InputFile();
+      i.path = stf.getPath().toString();
+      Range r = stf.getRange();
+      i.startRow = r.getStartKey() == null ? "null" : r.getStartKey().toString();
+      i.endRow = r.getEndKey() == null ? "null" : r.getEndKey().toString();
+      return i;
+    }
+
+    private StoredTabletFile to() {
+      Range r = new Range(startRow.equals("null") ? null : startRow,
+          endRow.equals("null") ? null : endRow);
+      return StoredTabletFile.of(new Path(path), r);
+    }
+  }
+
   // This class is used to serialize and deserialize this class using GSon. Any changes to this
   // class must consider persisted data.
   private static class GSonData {
-    String inputs;
+    List<InputFile> inputs;
     String tmp;
     String compactor;
     String kind;
@@ -112,35 +130,21 @@ private static class GSonData {
   }
 
   public String toJson() {
-
-    final Gson gson = GSON.get();
-
-    List<String> inputFiles =
-        jobFiles.stream().map(StoredTabletFile::getMetadata).collect(toList());
-    String inputFilesJson = gson.toJson(inputFiles);
-
     final GSonData jData = new GSonData();
-    jData.inputs = Base64.getEncoder().encodeToString(inputFilesJson.getBytes(UTF_8));
+    jData.inputs = jobFiles.stream().map(InputFile::from).collect(toList());
     jData.tmp = compactTmpName.insert().getMetadata();
     jData.compactor = compactorId;
     jData.kind = kind.name();
     jData.groupId = cgid.toString();
     jData.priority = priority;
     jData.propDels = propagateDeletes;
     jData.fateId = fateId == null ? null : fateId.canonical();
-    return gson.toJson(jData);
+    return GSON.get().toJson(jData);
   }
 
   public static CompactionMetadata fromJson(String json) {
-    final Gson gson = GSON.get();
-
-    GSonData jData = gson.fromJson(json, GSonData.class);
-
-    byte[] inputFilesDecoded = Base64.getDecoder().decode(jData.inputs);
-    List<String> inputFiles = gson.fromJson(new String(inputFilesDecoded, UTF_8),
-        new TypeToken<List<String>>() {}.getType());
-
-    return new CompactionMetadata(inputFiles.stream().map(StoredTabletFile::new).collect(toSet()),
+    GSonData jData = GSON.get().fromJson(json, GSonData.class);
+    return new CompactionMetadata(jData.inputs.stream().map(InputFile::to).collect(toSet()),
         StoredTabletFile.of(jData.tmp).getTabletFile(), jData.compactor,
         CompactionKind.valueOf(jData.kind), jData.priority, CompactorGroupId.of(jData.groupId),
         jData.propDels, jData.fateId == null ? null : FateId.from(jData.fateId));