nastra commented on code in PR #15713:
URL: https://github.com/apache/iceberg/pull/15713#discussion_r3527770656
##########
core/src/test/java/org/apache/iceberg/TestTables.java:
##########
@@ -410,4 +411,106 @@ public void deleteFiles(Iterable<String> pathsToDelete)
throws BulkDeletionFailu
throw new RuntimeException("Expected to mock this function");
}
}
+
+ /** A {@link FileIO} that enforces a limit on concurrent open input streams.
*/
+ public static class TrackingFileIO implements FileIO {
Review Comment:
the other alternative to testing this would be to have something like the
following, which wouldn't require introducing a separate FileIO class in
`TestTables` and we could keep everything isolated in a single test class:
```
@ExtendWith(ParameterizedTestExtension.class)
class TestManifestFilterManager extends TestBase {
@BeforeEach
public void before() {
LiveEntriesTracker.reset();
}
@TestTemplate
public void filterManifestsClosesLiveEntriesBeforeReopening() throws
IOException {
ExecutorService executor = Executors.newSingleThreadExecutor();
try {
ManifestFile manifest = writeManifest(1L, FILE_A);
TestDataFileFilterManager filterManager = new
TestDataFileFilterManager(executor);
filterManager.delete(FILE_A);
List<ManifestFile> filtered = filterManager.filterManifests(SCHEMA,
List.of(manifest));
assertThat(filtered).hasSize(1);
assertThat(filtered.get(0).path()).isNotEqualTo(manifest.path());
assertThat(LiveEntriesTracker.peakOpen())
.as("liveEntries() must be closed before opening again during
manifest rewrite")
.isEqualTo(1);
} finally {
executor.shutdownNow();
}
}
private static final class LiveEntriesTracker {
private static final AtomicInteger OPEN = new AtomicInteger(0);
private static final AtomicInteger PEAK = new AtomicInteger(0);
private LiveEntriesTracker() {}
static void reset() {
OPEN.set(0);
PEAK.set(0);
}
static int peakOpen() {
return PEAK.get();
}
static <T> CloseableIterable<T> track(CloseableIterable<T> iterable) {
PEAK.accumulateAndGet(OPEN.incrementAndGet(), Math::max);
return new CloseableIterable<T>() {
@Override
public void close() throws IOException {
iterable.close();
OPEN.decrementAndGet();
}
@Override
public CloseableIterator<T> iterator() {
return iterable.iterator();
}
};
}
}
private static class TrackingManifestReader extends
ManifestReader<DataFile> {
TrackingManifestReader(
ManifestFile manifest, FileIO io, Map<Integer, PartitionSpec>
specsById) {
super(
io.newInputFile(manifest.path()),
manifest.partitionSpecId(),
specsById,
InheritableMetadataFactory.fromManifest(manifest),
manifest.firstRowId(),
true,
FileType.DATA_FILES);
}
@Override
CloseableIterable<ManifestEntry<DataFile>> liveEntries() {
return LiveEntriesTracker.track(super.liveEntries());
}
}
private class TestDataFileFilterManager extends
ManifestFilterManager<DataFile> {
private final AtomicInteger manifestCounter = new AtomicInteger(0);
TestDataFileFilterManager(ExecutorService executor) {
super(table.specs(), (Supplier<ExecutorService>) () -> executor);
}
@Override
protected void deleteFile(String location) {
// filtering test only
}
@Override
protected ManifestWriter<DataFile> newManifestWriter(PartitionSpec spec)
{
String path =
String.format(
"%s/filtered-%d%s",
table.location(),
manifestCounter.incrementAndGet(),
manifestFormat().addExtension(""));
OutputFile outputFile = FILE_IO.newOutputFile(path);
return ManifestFiles.write(formatVersion, spec, outputFile, 1L);
}
@Override
protected ManifestReader<DataFile> newManifestReader(ManifestFile
manifest) {
return new TrackingManifestReader(manifest, FILE_IO, table.specs());
}
@Override
protected Set<DataFile> newFileSet() {
return DataFileSet.create();
}
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
