diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/DBScanner.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/DBScanner.java
index 0c38fbe33ba..4653aa3eeb3 100644
--- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/DBScanner.java
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/DBScanner.java
@@ -55,9 +55,11 @@
 import java.io.BufferedWriter;
 import java.io.IOException;
 import java.io.PrintWriter;
+import java.lang.reflect.Field;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -121,6 +123,11 @@ public class DBScanner implements Callable<Void>, SubcommandWithParent {
       description = "Key at which iteration of the DB ends")
   private String endKey;
 
+  @CommandLine.Option(names = {"--fields"},
+      description = "Comma-separated list of fields needed for each value. " +
+          "eg.) \"name,acls.type\" for showing name and type under acls.")
+  private String fieldsFilter;
+
   @CommandLine.Option(names = {"--dnSchema", "--dn-schema", "-d"},
       description = "Datanode DB Schema Version: V1/V2/V3",
       defaultValue = "V3")
@@ -291,7 +298,7 @@ private void processRecords(ManagedRocksIterator iterator,
         }
         Future<Void> future = threadPool.submit(
             new Task(dbColumnFamilyDef, batch, logWriter, sequenceId,
-                withKey, schemaV3));
+                withKey, schemaV3, fieldsFilter));
         futures.add(future);
         batch = new ArrayList<>(batchSize);
         sequenceId++;
@@ -299,7 +306,7 @@ private void processRecords(ManagedRocksIterator iterator,
     }
     if (!batch.isEmpty()) {
       Future<Void> future = threadPool.submit(new Task(dbColumnFamilyDef,
-          batch, logWriter, sequenceId, withKey, schemaV3));
+          batch, logWriter, sequenceId, withKey, schemaV3, fieldsFilter));
       futures.add(future);
     }
 
@@ -465,22 +472,51 @@ private static class Task implements Callable<Void> {
     private final long sequenceId;
     private final boolean withKey;
     private final boolean schemaV3;
+    private String valueFields;
 
     Task(DBColumnFamilyDefinition dbColumnFamilyDefinition,
          ArrayList<ByteArrayKeyValue> batch, LogWriter logWriter,
-         long sequenceId, boolean withKey, boolean schemaV3) {
+         long sequenceId, boolean withKey, boolean schemaV3, String valueFields) {
       this.dbColumnFamilyDefinition = dbColumnFamilyDefinition;
       this.batch = batch;
       this.logWriter = logWriter;
       this.sequenceId = sequenceId;
       this.withKey = withKey;
       this.schemaV3 = schemaV3;
+      this.valueFields = valueFields;
+    }
+
+    Map<String, Object> getFieldSplit(List<String> fields, Map<String, Object> fieldMap) {
+      int len = fields.size();
+      if (fieldMap == null) {
+        fieldMap = new HashMap<>();
+      }
+      if (len == 1) {
+        fieldMap.putIfAbsent(fields.get(0), null);
+      } else {
+        Map<String, Object> fieldMapGet = (Map<String, Object>) fieldMap.get(fields.get(0));
+        if (fieldMapGet == null) {
+          fieldMap.put(fields.get(0), getFieldSplit(fields.subList(1, len), null));
+        } else {
+          fieldMap.put(fields.get(0), getFieldSplit(fields.subList(1, len), fieldMapGet));
+        }
+      }
+      return fieldMap;
     }
 
     @Override
     public Void call() {
       try {
         ArrayList<String> results = new ArrayList<>(batch.size());
+        Map<String, Object> fieldsSplitMap = new HashMap<>();
+
+        if (valueFields != null) {
+          for (String field : valueFields.split(",")) {
+            String[] subfields = field.split("\\.");
+            fieldsSplitMap = getFieldSplit(Arrays.asList(subfields), fieldsSplitMap);
+          }
+        }
+
         for (ByteArrayKeyValue byteArrayKeyValue : batch) {
           StringBuilder sb = new StringBuilder();
           if (!(sequenceId == FIRST_SEQUENCE_ID && results.isEmpty())) {
@@ -515,16 +551,92 @@ public Void call() {
 
           Object o = dbColumnFamilyDefinition.getValueCodec()
               .fromPersistedFormat(byteArrayKeyValue.getValue());
-          sb.append(WRITER.writeValueAsString(o));
+
+          if (valueFields != null) {
+            Map<String, Object> filteredValue = new HashMap<>();
+            filteredValue.putAll(getFilteredObject(o, dbColumnFamilyDefinition.getValueType(), fieldsSplitMap));
+            sb.append(WRITER.writeValueAsString(filteredValue));
+          } else {
+            sb.append(WRITER.writeValueAsString(o));
+          }
+
           results.add(sb.toString());
         }
         logWriter.log(results, sequenceId);
-      } catch (Exception e) {
+      } catch (IOException e) {
         exception = true;
         LOG.error("Exception parse Object", e);
       }
       return null;
     }
+
+    Map<String, Object> getFilteredObject(Object obj, Class<?> clazz, Map<String, Object> fieldsSplitMap) {
+      Map<String, Object> valueMap = new HashMap<>();
+      for (Map.Entry<String, Object> field : fieldsSplitMap.entrySet()) {
+        try {
+          Field valueClassField = getRequiredFieldFromAllFields(clazz, field.getKey());
+          Object valueObject = valueClassField.get(obj);
+          Map<String, Object> subfields = (Map<String, Object>) field.getValue();
+
+          if (subfields == null) {
+            valueMap.put(field.getKey(), valueObject);
+          } else {
+            if (Collection.class.isAssignableFrom(valueObject.getClass())) {
+              List<Object> subfieldObjectsList =
+                  getFilteredObjectCollection((Collection) valueObject, subfields);
+              valueMap.put(field.getKey(), subfieldObjectsList);
+            } else if (Map.class.isAssignableFrom(valueObject.getClass())) {
+              Map<Object, Object> subfieldObjectsMap = new HashMap<>();
+              Map<?, ?> valueObjectMap = (Map<?, ?>) valueObject;
+              for (Map.Entry<?, ?> ob : valueObjectMap.entrySet()) {
+                Object subfieldValue;
+                if (Collection.class.isAssignableFrom(ob.getValue().getClass())) {
+                  subfieldValue = getFilteredObjectCollection((Collection)ob.getValue(), subfields);
+                } else {
+                  subfieldValue = getFilteredObject(ob.getValue(), ob.getValue().getClass(), subfields);
+                }
+                subfieldObjectsMap.put(ob.getKey(), subfieldValue);
+              }
+              valueMap.put(field.getKey(), subfieldObjectsMap);
+            } else {
+              valueMap.put(field.getKey(),
+                  getFilteredObject(valueObject, valueClassField.getType(), subfields));
+            }
+          }
+        } catch (NoSuchFieldException ex) {
+          err().println("ERROR: no such field: " + field);
+        } catch (IllegalAccessException e) {
+          err().println("ERROR: Cannot get field from object: " + field);
+        }
+      }
+      return valueMap;
+    }
+
+    List<Object> getFilteredObjectCollection(Collection<?> valueObject, Map<String, Object> fields)
+        throws NoSuchFieldException, IllegalAccessException {
+      List<Object> subfieldObjectsList = new ArrayList<>();
+      for (Object ob : valueObject) {
+        Object subfieldValue = getFilteredObject(ob, ob.getClass(), fields);
+        subfieldObjectsList.add(subfieldValue);
+      }
+      return subfieldObjectsList;
+    }
+
+    Field getRequiredFieldFromAllFields(Class clazz, String fieldName) throws NoSuchFieldException {
+      List<Field> classFieldList = ValueSchema.getAllFields(clazz);
+      Field classField = null;
+      for (Field f : classFieldList) {
+        if (f.getName().equals(fieldName)) {
+          classField = f;
+          break;
+        }
+      }
+      if (classField == null) {
+        throw new NoSuchFieldException();
+      }
+      classField.setAccessible(true);
+      return classField;
+    }
   }
 
   private static class ByteArrayKeyValue {
diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/ValueSchema.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/ValueSchema.java
index a5029b3e6b9..b06be2aff53 100644
--- a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/ValueSchema.java
+++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/ValueSchema.java
@@ -88,7 +88,7 @@ public Void call() throws Exception {
 
     String dbPath = parent.getDbPath();
     Map<String, Object> fields = new HashMap<>();
-    success = getValueFields(dbPath, fields);
+    success = getValueFields(dbPath, fields, depth, tableName, dnDBSchemaVersion);
 
     out().println(JsonUtils.toJsonStringWithDefaultPrettyPrinter(fields));
 
@@ -101,7 +101,8 @@ public Void call() throws Exception {
     return null;
   }
 
-  private boolean getValueFields(String dbPath, Map<String, Object> valueSchema) {
+  public static boolean getValueFields(String dbPath, Map<String, Object> valueSchema, int d, String table,
+                                       String dnDBSchemaVersion) {
 
     dbPath = removeTrailingSlashIfNeeded(dbPath);
     DBDefinitionFactory.setDnDBSchemaVersion(dnDBSchemaVersion);
@@ -111,19 +112,19 @@ private boolean getValueFields(String dbPath, Map<String, Object> valueSchema) {
       return false;
     }
     final DBColumnFamilyDefinition<?, ?> columnFamilyDefinition =
-        dbDefinition.getColumnFamily(tableName);
+        dbDefinition.getColumnFamily(table);
     if (columnFamilyDefinition == null) {
-      err().print("Error: Table with name '" + tableName + "' not found");
+      err().print("Error: Table with name '" + table + "' not found");
       return false;
     }
 
     Class<?> c = columnFamilyDefinition.getValueType();
-    valueSchema.put(c.getSimpleName(), getFieldsStructure(c, depth));
+    valueSchema.put(c.getSimpleName(), getFieldsStructure(c, d));
 
     return true;
   }
 
-  private Object getFieldsStructure(Class<?> clazz, int currentDepth) {
+  private static Object getFieldsStructure(Class<?> clazz, int currentDepth) {
     if (clazz.isPrimitive() || String.class.equals(clazz)) {
       return clazz.getSimpleName();
     } else if (currentDepth == 0) {
@@ -148,7 +149,7 @@ private Object getFieldsStructure(Class<?> clazz, int currentDepth) {
     }
   }
 
-  private List<Field> getAllFields(Class clazz) {
+  public static List<Field> getAllFields(Class clazz) {
     // NOTE: Schema of interface type, like ReplicationConfig, cannot be fetched.
     //       An empty list "[]" will be shown for such types of fields.
     if (clazz == null) {
@@ -176,7 +177,7 @@ public Class<?> getParentType() {
     return RDBParser.class;
   }
 
-  private String removeTrailingSlashIfNeeded(String dbPath) {
+  private static String removeTrailingSlashIfNeeded(String dbPath) {
     if (dbPath.endsWith(OzoneConsts.OZONE_URI_DELIMITER)) {
       dbPath = dbPath.substring(0, dbPath.length() - 1);
     }