@Test public void testIncrementalWithMultipleCommits() throws IOException { File partitionDir = InputFormatTestUtil.prepareTable(basePath, baseFileFormat, 10, "100"); createCommitFile(basePath, "100", "2016/05/01"); FileInputFormat.setInputPaths(jobConf, partitionDir.getPath()); InputFormatTestUtil.simulateUpdates(partitionDir, baseFileExtension, "100", 5, "200", false); createCommitFile(basePath, "200", "2016/05/01"); InputFormatTestUtil.simulateUpdates(partitionDir, baseFileExtension, "100", 4, "300", false); createCommitFile(basePath, "300", "2016/05/01"); InputFormatTestUtil.simulateUpdates(partitionDir, baseFileExtension, "100", 3, "400", false); createCommitFile(basePath, "400", "2016/05/01"); InputFormatTestUtil.simulateUpdates(partitionDir, baseFileExtension, "100", 2, "500", false); createCommitFile(basePath, "500", "2016/05/01"); InputFormatTestUtil.simulateUpdates(partitionDir, baseFileExtension, "100", 1, "600", false); createCommitFile(basePath, "600", "2016/05/01"); InputFormatTestUtil.setupIncremental(jobConf, "100", 1); FileStatus[] files = inputFormat.listStatus(jobConf); assertEquals(5, files.length, "Pulling 1 commit from 100, should get us the 5 files committed at 200"); ensureFilesInCommit("Pulling 1 commit from 100, should get us the 5 files committed at 200", files, "200", 5); InputFormatTestUtil.setupIncremental(jobConf, "100", 3); files = inputFormat.listStatus(jobConf); assertEquals(5, files.length, "Pulling 3 commits from 100, should get us the 3 files from 400 commit, 1 file from 300 " + "commit and 1 file from 200 commit"); ensureFilesInCommit("Pulling 3 commits from 100, should get us the 3 files from 400 commit", files, "400", 3); ensureFilesInCommit("Pulling 3 commits from 100, should get us the 1 files from 300 commit", files, "300", 1); ensureFilesInCommit("Pulling 3 commits from 100, should get us the 1 files from 200 commit", files, "200", 1); InputFormatTestUtil.setupIncremental(jobConf, "100", HoodieHiveUtils.MAX_COMMIT_ALL); files = inputFormat.listStatus(jobConf); assertEquals(5, files.length, "Pulling all commits from 100, should get us the 1 file from each of 200,300,400,500,400 commits"); ensureFilesInCommit("Pulling all commits from 100, should get us the 1 files from 600 commit", files, "600", 1); ensureFilesInCommit("Pulling all commits from 100, should get us the 1 files from 500 commit", files, "500", 1); ensureFilesInCommit("Pulling all commits from 100, should get us the 1 files from 400 commit", files, "400", 1); ensureFilesInCommit("Pulling all commits from 100, should get us the 1 files from 300 commit", files, "300", 1); ensureFilesInCommit("Pulling all commits from 100, should get us the 1 files from 200 commit", files, "200", 1); }
@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);  }