@Test public void testInputFormatLoad() throws IOException { File partitionDir = InputFormatTestUtil.prepareTable(basePath, baseFileFormat, 10, "100"); InputFormatTestUtil.commit(basePath, "100"); FileInputFormat.setInputPaths(jobConf, partitionDir.getPath()); InputSplit[] inputSplits = inputFormat.getSplits(jobConf, 10); assertEquals(10, inputSplits.length); FileStatus[] files = inputFormat.listStatus(jobConf); assertEquals(10, files.length); }
@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);  }