HDDS-10568. When the ldb command is executed, it is output by line

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/debug/TestLDBCli.java

@@ -19,6 +19,8 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParser;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.hdds.StringUtils;
 import org.apache.hadoop.hdds.client.BlockID;
@@ -52,6 +54,8 @@
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.io.FileInputStream;
+import java.io.InputStreamReader;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -68,6 +72,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 /**
  * This class tests `ozone debug ldb` CLI that reads from a RocksDB directory.
@@ -341,6 +346,52 @@ void testScanOfPipelinesWhenNoData() throws IOException {
     assertEquals("", stderr.toString());
   }
 
+  @Test
+  void testScanWithRecordsPerFile() throws IOException {
+    // Prepare dummy table
+    int recordsCount = 5;
+    prepareTable(KEY_TABLE, false, recordsCount);
+
+    String scanDir1 = tempDir.getAbsolutePath() + "/scandir1";
+    // Prepare scan args
+    int maxRecordsPerFile = 2;
+    List<String> completeScanArgs1 = new ArrayList<>(Arrays.asList(
+        "--db", dbStore.getDbLocation().getAbsolutePath(),
+        "scan",
+        "--column-family", KEY_TABLE, "--out", scanDir1 + File.separator + "keytable",
+        "--max-records-per-file", String.valueOf(maxRecordsPerFile)));
+    File tmpDir1 = new File(scanDir1);
+    tmpDir1.deleteOnExit();
+
+    int exitCode1 = cmd.execute(completeScanArgs1.toArray(new String[0]));
+    assertEquals(0, exitCode1);
+    assertTrue(tmpDir1.isDirectory());
+    File[] subFiles = tmpDir1.listFiles();
+    assertNotNull(subFiles);
+    assertEquals(Math.ceil(recordsCount / (maxRecordsPerFile * 1.0)), subFiles.length);
+    for (File subFile : subFiles) {
+      try (FileInputStream inputStream = new FileInputStream(subFile)) {
+        JsonElement jsonElement = JsonParser.parseReader(new InputStreamReader(inputStream, "UTF-8"));
+        assertTrue(jsonElement.isJsonArray() || jsonElement.isJsonObject());
+      }
+    }
+
+    String scanDir2 = tempDir.getAbsolutePath() + "/scandir2";
+    // Used with parameter '-l'
+    List<String> completeScanArgs2 = new ArrayList<>(Arrays.asList(
+        "--db", dbStore.getDbLocation().getAbsolutePath(),
+        "scan",
+        "--column-family", KEY_TABLE, "--out", scanDir2 + File.separator + "keytable",
+        "--max-records-per-file", String.valueOf(maxRecordsPerFile), "-l", "2"));
+    File tmpDir2 = new File(scanDir2);
+    tmpDir2.deleteOnExit();
+
+    int exitCode2 = cmd.execute(completeScanArgs2.toArray(new String[0]));
+    assertEquals(0, exitCode2);
+    assertTrue(tmpDir2.isDirectory());
+    assertEquals(1, tmpDir2.listFiles().length);
+  }
+
   @Test
   void testSchemaCommand() throws IOException {
     // Prepare dummy table
@@ -384,7 +435,7 @@ private void assertContents(Map<String, ?> expected, String actualStr)
    * @param tableName table name
    * @param schemaV3 set to true for SchemaV3. applicable to block_data table
    */
-  private void prepareTable(String tableName, boolean schemaV3)
+  private void prepareTable(String tableName, boolean schemaV3, int... recordsCount)
       throws IOException {
 
     switch (tableName) {
@@ -394,8 +445,12 @@ private void prepareTable(String tableName, boolean schemaV3)
           .setPath(tempDir.toPath()).addTable(KEY_TABLE).build();
 
       Table<byte[], byte[]> keyTable = dbStore.getTable(KEY_TABLE);
-      // Insert 5 keys
-      for (int i = 1; i <= 5; i++) {
+      // Insert the number of keys specified by recordsCount[0]
+      int count = 5;
+      if (recordsCount.length > 0) {
+        count = recordsCount[0];
+      }
+      for (int i = 1; i <= count; i++) {
         String key = "key" + i;
         OmKeyInfo value = OMRequestTestUtils.createOmKeyInfo("vol1", "buck1",
             key, ReplicationConfig.fromProtoTypeAndFactor(STAND_ALONE, HddsProtos.ReplicationFactor.ONE)).build();

hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/DBScanner.java

@@ -54,6 +54,7 @@
 import picocli.CommandLine;
 
 import java.io.BufferedWriter;
+import java.io.File;
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.lang.reflect.Field;
@@ -172,6 +173,14 @@ public class DBScanner implements Callable<Void>, SubcommandWithParent {
       defaultValue = "10")
   private int threadCount;
 
+  @CommandLine.Option(names = {"--max-records-per-file"},
+      description = "The number of records to print per file.",
+      defaultValue = "0")
+  private long recordsPerFile;
+
+  private int fileSuffix = 0;
+  private long globalCount = 0;
+
   private static final String KEY_SEPARATOR_SCHEMA_V3 =
       new OzoneConfiguration().getObject(DatanodeConfiguration.class)
           .getContainerSchemaV3KeySeparator();
@@ -180,7 +189,8 @@ public class DBScanner implements Callable<Void>, SubcommandWithParent {
 
   @Override
   public Void call() throws Exception {
-
+    fileSuffix = 0;
+    globalCount = 0;
     List<ColumnFamilyDescriptor> cfDescList =
         RocksDBUtils.getColumnFamilyDescriptors(parent.getDbPath());
     final List<ColumnFamilyHandle> cfHandleList = new ArrayList<>();
@@ -240,11 +250,31 @@ private boolean displayTable(ManagedRocksIterator iterator,
       return displayTable(iterator, dbColumnFamilyDef, out(), schemaV3);
     }
 
+    // If there are no parent directories, create them
+    if (recordsPerFile > 0) {
+      File file = new File(fileName);
+      File parentFile = file.getParentFile();
+      if (!parentFile.exists()) {
+        boolean flg = parentFile.mkdirs();
+        if (!flg) {
+          throw new IOException("An exception occurred while creating " +
+              "the directory. Directorys: " + parentFile.getAbsolutePath());
+        }
+      }
+    }
+
     // Write to file output
-    try (PrintWriter out = new PrintWriter(new BufferedWriter(
-        new PrintWriter(fileName, UTF_8.name())))) {
-      return displayTable(iterator, dbColumnFamilyDef, out, schemaV3);
+    while (iterator.get().isValid() && withinLimit(globalCount)) {
+      String fileNameTarget = recordsPerFile > 0 ? fileName + "." + fileSuffix++ :
+          fileName;
+      try (PrintWriter out = new PrintWriter(new BufferedWriter(
+          new PrintWriter(fileNameTarget, UTF_8.name())))) {
+        if (!displayTable(iterator, dbColumnFamilyDef, out, schemaV3)) {
+          return false;
+        }
+      }
     }
+    return true;
   }
 
   private boolean displayTable(ManagedRocksIterator iterator,
@@ -314,7 +344,7 @@ private void processRecords(ManagedRocksIterator iterator,
       }
     }
 
-    while (withinLimit(count) && iterator.get().isValid() && !exception && !reachedEnd) {
+    while (withinLimit(globalCount) && iterator.get().isValid() && !exception && !reachedEnd) {
       // if invalid endKey is given, it is ignored
       if (null != endKey && Arrays.equals(iterator.get().key(), getValueObject(dbColumnFamilyDef, endKey))) {
         reachedEnd = true;
@@ -326,6 +356,7 @@ private void processRecords(ManagedRocksIterator iterator,
         // the record passes the filter
         batch.add(new ByteArrayKeyValue(
             iterator.get().key(), iterator.get().value()));
+        globalCount++;
         count++;
         if (batch.size() >= batchSize) {
           while (logWriter.getInflightLogCount() > threadCount * 10L
@@ -343,6 +374,9 @@ private void processRecords(ManagedRocksIterator iterator,
         }
       }
       iterator.get().next();
+      if ((recordsPerFile > 0) && (count >= recordsPerFile)) {
+        break;
+      }
     }
     if (!batch.isEmpty()) {
       Future<Void> future = threadPool.submit(new Task(dbColumnFamilyDef,