@Test public void testShowLogFileRecords() throws IOException, URISyntaxException { CommandResult cr = getShell().executeCommand("show logfile records --logFilePathPattern " + partitionPath + "/*"); assertTrue(cr.isSuccess()); List<IndexedRecord> records = SchemaTestUtil.generateTestRecords(0, 10); String[][] rows = records.stream().map(r -> new String[]{r.toString()}).toArray(String[][]::new); String expected = HoodiePrintHelper.print(new String[] {HoodieTableHeaderFields.HEADER_RECORDS}, rows); expected = removeNonWordAndStripSpace(expected); String got = removeNonWordAndStripSpace(cr.getResult().toString()); assertEquals(expected, got); }
@CliCommand(value = "show logfile records", help = "Read records from log files") public String showLogFileRecords( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "10") final Integer limit, @CliOption(key = "logFilePathPattern", mandatory = true, help = "Fully qualified paths for the log files") final String logFilePathPattern, @CliOption(key = "mergeRecords", help = "If the records in the log files should be merged", unspecifiedDefaultValue = "false") final Boolean shouldMerge) throws IOException { System.out.println("===============> Showing only " + limit + " records <==============="); HoodieTableMetaClient client = HoodieCLI.getTableMetaClient(); FileSystem fs = client.getFs(); List<String> logFilePaths = FSUtils.getGlobStatusExcludingMetaFolder(fs, new Path(logFilePathPattern)).stream() .map(status -> status.getPath().toString()).sorted(Comparator.reverseOrder()) .collect(Collectors.toList()); assert logFilePaths.size() > 0 : "There is no log file"; AvroSchemaConverter converter = new AvroSchemaConverter(); Schema readerSchema = converter.convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(fs, new Path(logFilePaths.get(logFilePaths.size() - 1))))); List<IndexedRecord> allRecords = new ArrayList<>(); if (shouldMerge) { System.out.println("===========================> MERGING RECORDS <==================="); HoodieMergedLogRecordScanner scanner = new HoodieMergedLogRecordScanner(fs, client.getBasePath(), logFilePaths, readerSchema, client.getActiveTimeline().getCommitTimeline().lastInstant().get().getTimestamp(), HoodieMemoryConfig.DEFAULT_MAX_MEMORY_FOR_SPILLABLE_MAP_IN_BYTES, Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_LAZY_BLOCK_READ_ENABLED), Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_REVERSE_LOG_READ_ENABLED), HoodieMemoryConfig.DEFAULT_MAX_DFS_STREAM_BUFFER_SIZE, HoodieMemoryConfig.DEFAULT_SPILLABLE_MAP_BASE_PATH); for (HoodieRecord<? extends HoodieRecordPayload> hoodieRecord : scanner) { Option<IndexedRecord> record = hoodieRecord.getData().getInsertValue(readerSchema); if (allRecords.size() < limit) { allRecords.add(record.get()); } } } else { for (String logFile : logFilePaths) { Schema writerSchema = new AvroSchemaConverter() .convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(client.getFs(), new Path(logFile)))); HoodieLogFormat.Reader reader = HoodieLogFormat.newReader(fs, new HoodieLogFile(new Path(logFile)), writerSchema); while (reader.hasNext()) { HoodieLogBlock n = reader.next(); if (n instanceof HoodieDataBlock) { HoodieDataBlock blk = (HoodieDataBlock) n; List<IndexedRecord> records = blk.getRecords(); for (IndexedRecord record : records) { if (allRecords.size() < limit) { allRecords.add(record); } } } } reader.close(); if (allRecords.size() >= limit) { break; } } } String[][] rows = new String[allRecords.size()][]; int i = 0; for (IndexedRecord record : allRecords) { String[] data = new String[1]; data[0] = record.toString(); rows[i] = data; i++; } return HoodiePrintHelper.print(new String[] {HoodieTableHeaderFields.HEADER_RECORDS}, rows); }
HoodieLogFileCommand implements CommandMarker { @CliCommand(value = "show logfile records", help = "Read records from log files") public String showLogFileRecords( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "10") final Integer limit, @CliOption(key = "logFilePathPattern", mandatory = true, help = "Fully qualified paths for the log files") final String logFilePathPattern, @CliOption(key = "mergeRecords", help = "If the records in the log files should be merged", unspecifiedDefaultValue = "false") final Boolean shouldMerge) throws IOException { System.out.println("===============> Showing only " + limit + " records <==============="); HoodieTableMetaClient client = HoodieCLI.getTableMetaClient(); FileSystem fs = client.getFs(); List<String> logFilePaths = FSUtils.getGlobStatusExcludingMetaFolder(fs, new Path(logFilePathPattern)).stream() .map(status -> status.getPath().toString()).sorted(Comparator.reverseOrder()) .collect(Collectors.toList()); assert logFilePaths.size() > 0 : "There is no log file"; AvroSchemaConverter converter = new AvroSchemaConverter(); Schema readerSchema = converter.convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(fs, new Path(logFilePaths.get(logFilePaths.size() - 1))))); List<IndexedRecord> allRecords = new ArrayList<>(); if (shouldMerge) { System.out.println("===========================> MERGING RECORDS <==================="); HoodieMergedLogRecordScanner scanner = new HoodieMergedLogRecordScanner(fs, client.getBasePath(), logFilePaths, readerSchema, client.getActiveTimeline().getCommitTimeline().lastInstant().get().getTimestamp(), HoodieMemoryConfig.DEFAULT_MAX_MEMORY_FOR_SPILLABLE_MAP_IN_BYTES, Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_LAZY_BLOCK_READ_ENABLED), Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_REVERSE_LOG_READ_ENABLED), HoodieMemoryConfig.DEFAULT_MAX_DFS_STREAM_BUFFER_SIZE, HoodieMemoryConfig.DEFAULT_SPILLABLE_MAP_BASE_PATH); for (HoodieRecord<? extends HoodieRecordPayload> hoodieRecord : scanner) { Option<IndexedRecord> record = hoodieRecord.getData().getInsertValue(readerSchema); if (allRecords.size() < limit) { allRecords.add(record.get()); } } } else { for (String logFile : logFilePaths) { Schema writerSchema = new AvroSchemaConverter() .convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(client.getFs(), new Path(logFile)))); HoodieLogFormat.Reader reader = HoodieLogFormat.newReader(fs, new HoodieLogFile(new Path(logFile)), writerSchema); while (reader.hasNext()) { HoodieLogBlock n = reader.next(); if (n instanceof HoodieDataBlock) { HoodieDataBlock blk = (HoodieDataBlock) n; List<IndexedRecord> records = blk.getRecords(); for (IndexedRecord record : records) { if (allRecords.size() < limit) { allRecords.add(record); } } } } reader.close(); if (allRecords.size() >= limit) { break; } } } String[][] rows = new String[allRecords.size()][]; int i = 0; for (IndexedRecord record : allRecords) { String[] data = new String[1]; data[0] = record.toString(); rows[i] = data; i++; } return HoodiePrintHelper.print(new String[] {HoodieTableHeaderFields.HEADER_RECORDS}, rows); } }
HoodieLogFileCommand implements CommandMarker { @CliCommand(value = "show logfile records", help = "Read records from log files") public String showLogFileRecords( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "10") final Integer limit, @CliOption(key = "logFilePathPattern", mandatory = true, help = "Fully qualified paths for the log files") final String logFilePathPattern, @CliOption(key = "mergeRecords", help = "If the records in the log files should be merged", unspecifiedDefaultValue = "false") final Boolean shouldMerge) throws IOException { System.out.println("===============> Showing only " + limit + " records <==============="); HoodieTableMetaClient client = HoodieCLI.getTableMetaClient(); FileSystem fs = client.getFs(); List<String> logFilePaths = FSUtils.getGlobStatusExcludingMetaFolder(fs, new Path(logFilePathPattern)).stream() .map(status -> status.getPath().toString()).sorted(Comparator.reverseOrder()) .collect(Collectors.toList()); assert logFilePaths.size() > 0 : "There is no log file"; AvroSchemaConverter converter = new AvroSchemaConverter(); Schema readerSchema = converter.convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(fs, new Path(logFilePaths.get(logFilePaths.size() - 1))))); List<IndexedRecord> allRecords = new ArrayList<>(); if (shouldMerge) { System.out.println("===========================> MERGING RECORDS <==================="); HoodieMergedLogRecordScanner scanner = new HoodieMergedLogRecordScanner(fs, client.getBasePath(), logFilePaths, readerSchema, client.getActiveTimeline().getCommitTimeline().lastInstant().get().getTimestamp(), HoodieMemoryConfig.DEFAULT_MAX_MEMORY_FOR_SPILLABLE_MAP_IN_BYTES, Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_LAZY_BLOCK_READ_ENABLED), Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_REVERSE_LOG_READ_ENABLED), HoodieMemoryConfig.DEFAULT_MAX_DFS_STREAM_BUFFER_SIZE, HoodieMemoryConfig.DEFAULT_SPILLABLE_MAP_BASE_PATH); for (HoodieRecord<? extends HoodieRecordPayload> hoodieRecord : scanner) { Option<IndexedRecord> record = hoodieRecord.getData().getInsertValue(readerSchema); if (allRecords.size() < limit) { allRecords.add(record.get()); } } } else { for (String logFile : logFilePaths) { Schema writerSchema = new AvroSchemaConverter() .convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(client.getFs(), new Path(logFile)))); HoodieLogFormat.Reader reader = HoodieLogFormat.newReader(fs, new HoodieLogFile(new Path(logFile)), writerSchema); while (reader.hasNext()) { HoodieLogBlock n = reader.next(); if (n instanceof HoodieDataBlock) { HoodieDataBlock blk = (HoodieDataBlock) n; List<IndexedRecord> records = blk.getRecords(); for (IndexedRecord record : records) { if (allRecords.size() < limit) { allRecords.add(record); } } } } reader.close(); if (allRecords.size() >= limit) { break; } } } String[][] rows = new String[allRecords.size()][]; int i = 0; for (IndexedRecord record : allRecords) { String[] data = new String[1]; data[0] = record.toString(); rows[i] = data; i++; } return HoodiePrintHelper.print(new String[] {HoodieTableHeaderFields.HEADER_RECORDS}, rows); }  }
HoodieLogFileCommand implements CommandMarker { @CliCommand(value = "show logfile records", help = "Read records from log files") public String showLogFileRecords( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "10") final Integer limit, @CliOption(key = "logFilePathPattern", mandatory = true, help = "Fully qualified paths for the log files") final String logFilePathPattern, @CliOption(key = "mergeRecords", help = "If the records in the log files should be merged", unspecifiedDefaultValue = "false") final Boolean shouldMerge) throws IOException { System.out.println("===============> Showing only " + limit + " records <==============="); HoodieTableMetaClient client = HoodieCLI.getTableMetaClient(); FileSystem fs = client.getFs(); List<String> logFilePaths = FSUtils.getGlobStatusExcludingMetaFolder(fs, new Path(logFilePathPattern)).stream() .map(status -> status.getPath().toString()).sorted(Comparator.reverseOrder()) .collect(Collectors.toList()); assert logFilePaths.size() > 0 : "There is no log file"; AvroSchemaConverter converter = new AvroSchemaConverter(); Schema readerSchema = converter.convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(fs, new Path(logFilePaths.get(logFilePaths.size() - 1))))); List<IndexedRecord> allRecords = new ArrayList<>(); if (shouldMerge) { System.out.println("===========================> MERGING RECORDS <==================="); HoodieMergedLogRecordScanner scanner = new HoodieMergedLogRecordScanner(fs, client.getBasePath(), logFilePaths, readerSchema, client.getActiveTimeline().getCommitTimeline().lastInstant().get().getTimestamp(), HoodieMemoryConfig.DEFAULT_MAX_MEMORY_FOR_SPILLABLE_MAP_IN_BYTES, Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_LAZY_BLOCK_READ_ENABLED), Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_REVERSE_LOG_READ_ENABLED), HoodieMemoryConfig.DEFAULT_MAX_DFS_STREAM_BUFFER_SIZE, HoodieMemoryConfig.DEFAULT_SPILLABLE_MAP_BASE_PATH); for (HoodieRecord<? extends HoodieRecordPayload> hoodieRecord : scanner) { Option<IndexedRecord> record = hoodieRecord.getData().getInsertValue(readerSchema); if (allRecords.size() < limit) { allRecords.add(record.get()); } } } else { for (String logFile : logFilePaths) { Schema writerSchema = new AvroSchemaConverter() .convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(client.getFs(), new Path(logFile)))); HoodieLogFormat.Reader reader = HoodieLogFormat.newReader(fs, new HoodieLogFile(new Path(logFile)), writerSchema); while (reader.hasNext()) { HoodieLogBlock n = reader.next(); if (n instanceof HoodieDataBlock) { HoodieDataBlock blk = (HoodieDataBlock) n; List<IndexedRecord> records = blk.getRecords(); for (IndexedRecord record : records) { if (allRecords.size() < limit) { allRecords.add(record); } } } } reader.close(); if (allRecords.size() >= limit) { break; } } } String[][] rows = new String[allRecords.size()][]; int i = 0; for (IndexedRecord record : allRecords) { String[] data = new String[1]; data[0] = record.toString(); rows[i] = data; i++; } return HoodiePrintHelper.print(new String[] {HoodieTableHeaderFields.HEADER_RECORDS}, rows); }  @CliCommand(value = "show logfile metadata", help = "Read commit metadata from log files") String showLogFileCommits(
      @CliOption(key = "logFilePathPattern", mandatory = true,
          help = "Fully qualified path for the log file") final String logFilePathPattern,
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); @CliCommand(value = "show logfile records", help = "Read records from log files") String showLogFileRecords(
      @CliOption(key = {"limit"}, help = "Limit commits",
          unspecifiedDefaultValue = "10") final Integer limit,
      @CliOption(key = "logFilePathPattern", mandatory = true,
          help = "Fully qualified paths for the log files") final String logFilePathPattern,
      @CliOption(key = "mergeRecords", help = "If the records in the log files should be merged",
          unspecifiedDefaultValue = "false") final Boolean shouldMerge); }
HoodieLogFileCommand implements CommandMarker { @CliCommand(value = "show logfile records", help = "Read records from log files") public String showLogFileRecords( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "10") final Integer limit, @CliOption(key = "logFilePathPattern", mandatory = true, help = "Fully qualified paths for the log files") final String logFilePathPattern, @CliOption(key = "mergeRecords", help = "If the records in the log files should be merged", unspecifiedDefaultValue = "false") final Boolean shouldMerge) throws IOException { System.out.println("===============> Showing only " + limit + " records <==============="); HoodieTableMetaClient client = HoodieCLI.getTableMetaClient(); FileSystem fs = client.getFs(); List<String> logFilePaths = FSUtils.getGlobStatusExcludingMetaFolder(fs, new Path(logFilePathPattern)).stream() .map(status -> status.getPath().toString()).sorted(Comparator.reverseOrder()) .collect(Collectors.toList()); assert logFilePaths.size() > 0 : "There is no log file"; AvroSchemaConverter converter = new AvroSchemaConverter(); Schema readerSchema = converter.convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(fs, new Path(logFilePaths.get(logFilePaths.size() - 1))))); List<IndexedRecord> allRecords = new ArrayList<>(); if (shouldMerge) { System.out.println("===========================> MERGING RECORDS <==================="); HoodieMergedLogRecordScanner scanner = new HoodieMergedLogRecordScanner(fs, client.getBasePath(), logFilePaths, readerSchema, client.getActiveTimeline().getCommitTimeline().lastInstant().get().getTimestamp(), HoodieMemoryConfig.DEFAULT_MAX_MEMORY_FOR_SPILLABLE_MAP_IN_BYTES, Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_LAZY_BLOCK_READ_ENABLED), Boolean.parseBoolean(HoodieCompactionConfig.DEFAULT_COMPACTION_REVERSE_LOG_READ_ENABLED), HoodieMemoryConfig.DEFAULT_MAX_DFS_STREAM_BUFFER_SIZE, HoodieMemoryConfig.DEFAULT_SPILLABLE_MAP_BASE_PATH); for (HoodieRecord<? extends HoodieRecordPayload> hoodieRecord : scanner) { Option<IndexedRecord> record = hoodieRecord.getData().getInsertValue(readerSchema); if (allRecords.size() < limit) { allRecords.add(record.get()); } } } else { for (String logFile : logFilePaths) { Schema writerSchema = new AvroSchemaConverter() .convert(Objects.requireNonNull(TableSchemaResolver.readSchemaFromLogFile(client.getFs(), new Path(logFile)))); HoodieLogFormat.Reader reader = HoodieLogFormat.newReader(fs, new HoodieLogFile(new Path(logFile)), writerSchema); while (reader.hasNext()) { HoodieLogBlock n = reader.next(); if (n instanceof HoodieDataBlock) { HoodieDataBlock blk = (HoodieDataBlock) n; List<IndexedRecord> records = blk.getRecords(); for (IndexedRecord record : records) { if (allRecords.size() < limit) { allRecords.add(record); } } } } reader.close(); if (allRecords.size() >= limit) { break; } } } String[][] rows = new String[allRecords.size()][]; int i = 0; for (IndexedRecord record : allRecords) { String[] data = new String[1]; data[0] = record.toString(); rows[i] = data; i++; } return HoodiePrintHelper.print(new String[] {HoodieTableHeaderFields.HEADER_RECORDS}, rows); }  @CliCommand(value = "show logfile metadata", help = "Read commit metadata from log files") String showLogFileCommits(
      @CliOption(key = "logFilePathPattern", mandatory = true,
          help = "Fully qualified path for the log file") final String logFilePathPattern,
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); @CliCommand(value = "show logfile records", help = "Read records from log files") String showLogFileRecords(
      @CliOption(key = {"limit"}, help = "Limit commits",
          unspecifiedDefaultValue = "10") final Integer limit,
      @CliOption(key = "logFilePathPattern", mandatory = true,
          help = "Fully qualified paths for the log files") final String logFilePathPattern,
      @CliOption(key = "mergeRecords", help = "If the records in the log files should be merged",
          unspecifiedDefaultValue = "false") final Boolean shouldMerge);  }