@Test public void testInputFormatUpdates() throws IOException { File partitionDir = InputFormatTestUtil.prepareTable(basePath, baseFileFormat, 10, "100"); InputFormatTestUtil.commit(basePath, "100"); FileInputFormat.setInputPaths(jobConf, partitionDir.getPath()); FileStatus[] files = inputFormat.listStatus(jobConf); assertEquals(10, files.length); InputFormatTestUtil.simulateUpdates(partitionDir, baseFileExtension, "100", 5, "200", true); files = inputFormat.listStatus(jobConf); assertEquals(10, files.length); ensureFilesInCommit("Commit 200 has not been committed. We should not see files from this commit", files, "200", 0); InputFormatTestUtil.commit(basePath, "200"); files = inputFormat.listStatus(jobConf); assertEquals(10, files.length); ensureFilesInCommit("5 files have been updated to commit 200. We should see 5 files from commit 200 and 5 " + "files from 100 commit", files, "200", 5); ensureFilesInCommit("5 files have been updated to commit 200. We should see 5 files from commit 100 and 5 " + "files from 200 commit", files, "100", 5); }
@Override public FileStatus[] listStatus(JobConf job) throws IOException { List<String> incrementalTables = HoodieHiveUtils.getIncrementalTableNames(Job.getInstance(job)); InputPathHandler inputPathHandler = new InputPathHandler(conf, getInputPaths(job), incrementalTables); List<FileStatus> returns = new ArrayList<>(); Map<String, HoodieTableMetaClient> tableMetaClientMap = inputPathHandler.getTableMetaClientMap(); for (String table : incrementalTables) { HoodieTableMetaClient metaClient = tableMetaClientMap.get(table); if (metaClient == null) { continue; } List<Path> inputPaths = inputPathHandler.getGroupedIncrementalPaths().get(metaClient); List<FileStatus> result = listStatusForIncrementalMode(job, metaClient, inputPaths); if (result != null) { returns.addAll(result); } } List<Path> nonHoodiePaths = inputPathHandler.getNonHoodieInputPaths(); if (nonHoodiePaths.size() > 0) { setInputPaths(job, nonHoodiePaths.toArray(new Path[nonHoodiePaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); returns.addAll(Arrays.asList(fileStatuses)); } List<Path> snapshotPaths = inputPathHandler.getSnapshotPaths(); if (snapshotPaths.size() > 0) { setInputPaths(job, snapshotPaths.toArray(new Path[snapshotPaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); Map<HoodieTableMetaClient, List<FileStatus>> groupedFileStatus = HoodieInputFormatUtils.groupFileStatusForSnapshotPaths(fileStatuses, HoodieFileFormat.PARQUET.getFileExtension(), tableMetaClientMap.values()); LOG.info("Found a total of " + groupedFileStatus.size() + " groups"); for (Map.Entry<HoodieTableMetaClient, List<FileStatus>> entry : groupedFileStatus.entrySet()) { List<FileStatus> result = HoodieInputFormatUtils.filterFileStatusForSnapshotMode(job, entry.getKey(), entry.getValue()); if (result != null) { returns.addAll(result); } } } return returns.toArray(new FileStatus[returns.size()]); }
HoodieParquetInputFormat extends MapredParquetInputFormat implements Configurable { @Override public FileStatus[] listStatus(JobConf job) throws IOException { List<String> incrementalTables = HoodieHiveUtils.getIncrementalTableNames(Job.getInstance(job)); InputPathHandler inputPathHandler = new InputPathHandler(conf, getInputPaths(job), incrementalTables); List<FileStatus> returns = new ArrayList<>(); Map<String, HoodieTableMetaClient> tableMetaClientMap = inputPathHandler.getTableMetaClientMap(); for (String table : incrementalTables) { HoodieTableMetaClient metaClient = tableMetaClientMap.get(table); if (metaClient == null) { continue; } List<Path> inputPaths = inputPathHandler.getGroupedIncrementalPaths().get(metaClient); List<FileStatus> result = listStatusForIncrementalMode(job, metaClient, inputPaths); if (result != null) { returns.addAll(result); } } List<Path> nonHoodiePaths = inputPathHandler.getNonHoodieInputPaths(); if (nonHoodiePaths.size() > 0) { setInputPaths(job, nonHoodiePaths.toArray(new Path[nonHoodiePaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); returns.addAll(Arrays.asList(fileStatuses)); } List<Path> snapshotPaths = inputPathHandler.getSnapshotPaths(); if (snapshotPaths.size() > 0) { setInputPaths(job, snapshotPaths.toArray(new Path[snapshotPaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); Map<HoodieTableMetaClient, List<FileStatus>> groupedFileStatus = HoodieInputFormatUtils.groupFileStatusForSnapshotPaths(fileStatuses, HoodieFileFormat.PARQUET.getFileExtension(), tableMetaClientMap.values()); LOG.info("Found a total of " + groupedFileStatus.size() + " groups"); for (Map.Entry<HoodieTableMetaClient, List<FileStatus>> entry : groupedFileStatus.entrySet()) { List<FileStatus> result = HoodieInputFormatUtils.filterFileStatusForSnapshotMode(job, entry.getKey(), entry.getValue()); if (result != null) { returns.addAll(result); } } } return returns.toArray(new FileStatus[returns.size()]); } }
HoodieParquetInputFormat extends MapredParquetInputFormat implements Configurable { @Override public FileStatus[] listStatus(JobConf job) throws IOException { List<String> incrementalTables = HoodieHiveUtils.getIncrementalTableNames(Job.getInstance(job)); InputPathHandler inputPathHandler = new InputPathHandler(conf, getInputPaths(job), incrementalTables); List<FileStatus> returns = new ArrayList<>(); Map<String, HoodieTableMetaClient> tableMetaClientMap = inputPathHandler.getTableMetaClientMap(); for (String table : incrementalTables) { HoodieTableMetaClient metaClient = tableMetaClientMap.get(table); if (metaClient == null) { continue; } List<Path> inputPaths = inputPathHandler.getGroupedIncrementalPaths().get(metaClient); List<FileStatus> result = listStatusForIncrementalMode(job, metaClient, inputPaths); if (result != null) { returns.addAll(result); } } List<Path> nonHoodiePaths = inputPathHandler.getNonHoodieInputPaths(); if (nonHoodiePaths.size() > 0) { setInputPaths(job, nonHoodiePaths.toArray(new Path[nonHoodiePaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); returns.addAll(Arrays.asList(fileStatuses)); } List<Path> snapshotPaths = inputPathHandler.getSnapshotPaths(); if (snapshotPaths.size() > 0) { setInputPaths(job, snapshotPaths.toArray(new Path[snapshotPaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); Map<HoodieTableMetaClient, List<FileStatus>> groupedFileStatus = HoodieInputFormatUtils.groupFileStatusForSnapshotPaths(fileStatuses, HoodieFileFormat.PARQUET.getFileExtension(), tableMetaClientMap.values()); LOG.info("Found a total of " + groupedFileStatus.size() + " groups"); for (Map.Entry<HoodieTableMetaClient, List<FileStatus>> entry : groupedFileStatus.entrySet()) { List<FileStatus> result = HoodieInputFormatUtils.filterFileStatusForSnapshotMode(job, entry.getKey(), entry.getValue()); if (result != null) { returns.addAll(result); } } } return returns.toArray(new FileStatus[returns.size()]); }  }
HoodieParquetInputFormat extends MapredParquetInputFormat implements Configurable { @Override public FileStatus[] listStatus(JobConf job) throws IOException { List<String> incrementalTables = HoodieHiveUtils.getIncrementalTableNames(Job.getInstance(job)); InputPathHandler inputPathHandler = new InputPathHandler(conf, getInputPaths(job), incrementalTables); List<FileStatus> returns = new ArrayList<>(); Map<String, HoodieTableMetaClient> tableMetaClientMap = inputPathHandler.getTableMetaClientMap(); for (String table : incrementalTables) { HoodieTableMetaClient metaClient = tableMetaClientMap.get(table); if (metaClient == null) { continue; } List<Path> inputPaths = inputPathHandler.getGroupedIncrementalPaths().get(metaClient); List<FileStatus> result = listStatusForIncrementalMode(job, metaClient, inputPaths); if (result != null) { returns.addAll(result); } } List<Path> nonHoodiePaths = inputPathHandler.getNonHoodieInputPaths(); if (nonHoodiePaths.size() > 0) { setInputPaths(job, nonHoodiePaths.toArray(new Path[nonHoodiePaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); returns.addAll(Arrays.asList(fileStatuses)); } List<Path> snapshotPaths = inputPathHandler.getSnapshotPaths(); if (snapshotPaths.size() > 0) { setInputPaths(job, snapshotPaths.toArray(new Path[snapshotPaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); Map<HoodieTableMetaClient, List<FileStatus>> groupedFileStatus = HoodieInputFormatUtils.groupFileStatusForSnapshotPaths(fileStatuses, HoodieFileFormat.PARQUET.getFileExtension(), tableMetaClientMap.values()); LOG.info("Found a total of " + groupedFileStatus.size() + " groups"); for (Map.Entry<HoodieTableMetaClient, List<FileStatus>> entry : groupedFileStatus.entrySet()) { List<FileStatus> result = HoodieInputFormatUtils.filterFileStatusForSnapshotMode(job, entry.getKey(), entry.getValue()); if (result != null) { returns.addAll(result); } } } return returns.toArray(new FileStatus[returns.size()]); }  @Override FileStatus[] listStatus(JobConf job); void setConf(Configuration conf); @Override Configuration getConf(); @Override RecordReader<NullWritable, ArrayWritable> getRecordReader(final InputSplit split, final JobConf job,
                                                                   final Reporter reporter); }
HoodieParquetInputFormat extends MapredParquetInputFormat implements Configurable { @Override public FileStatus[] listStatus(JobConf job) throws IOException { List<String> incrementalTables = HoodieHiveUtils.getIncrementalTableNames(Job.getInstance(job)); InputPathHandler inputPathHandler = new InputPathHandler(conf, getInputPaths(job), incrementalTables); List<FileStatus> returns = new ArrayList<>(); Map<String, HoodieTableMetaClient> tableMetaClientMap = inputPathHandler.getTableMetaClientMap(); for (String table : incrementalTables) { HoodieTableMetaClient metaClient = tableMetaClientMap.get(table); if (metaClient == null) { continue; } List<Path> inputPaths = inputPathHandler.getGroupedIncrementalPaths().get(metaClient); List<FileStatus> result = listStatusForIncrementalMode(job, metaClient, inputPaths); if (result != null) { returns.addAll(result); } } List<Path> nonHoodiePaths = inputPathHandler.getNonHoodieInputPaths(); if (nonHoodiePaths.size() > 0) { setInputPaths(job, nonHoodiePaths.toArray(new Path[nonHoodiePaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); returns.addAll(Arrays.asList(fileStatuses)); } List<Path> snapshotPaths = inputPathHandler.getSnapshotPaths(); if (snapshotPaths.size() > 0) { setInputPaths(job, snapshotPaths.toArray(new Path[snapshotPaths.size()])); FileStatus[] fileStatuses = super.listStatus(job); Map<HoodieTableMetaClient, List<FileStatus>> groupedFileStatus = HoodieInputFormatUtils.groupFileStatusForSnapshotPaths(fileStatuses, HoodieFileFormat.PARQUET.getFileExtension(), tableMetaClientMap.values()); LOG.info("Found a total of " + groupedFileStatus.size() + " groups"); for (Map.Entry<HoodieTableMetaClient, List<FileStatus>> entry : groupedFileStatus.entrySet()) { List<FileStatus> result = HoodieInputFormatUtils.filterFileStatusForSnapshotMode(job, entry.getKey(), entry.getValue()); if (result != null) { returns.addAll(result); } } } return returns.toArray(new FileStatus[returns.size()]); }  @Override FileStatus[] listStatus(JobConf job); void setConf(Configuration conf); @Override Configuration getConf(); @Override RecordReader<NullWritable, ArrayWritable> getRecordReader(final InputSplit split, final JobConf job,
                                                                   final Reporter reporter);  }