feat: DH-18174: Delay reading from parquet file when creating table and column location

[malhotrashivam]

https://deephaven.atlassian.net/browse/DH-18174

[rcaudy]

Please double check that we're not going to NPE because of a missed initialization.

[rcaudy]

.

[rcaudy]

I'm concerned about a potential side effect of this change, but I think we're OK. @malhotrashivam please confirm you've tested coalescing a table with many Parquet files that doesn't use _metadata and _common_metadata files, and that there is no performance regression there.

SourceTable has the following:

    private void maybeAddLocations(@NotNull final Collection<LiveSupplier<ImmutableTableLocationKey>> locationKeys) {
        if (locationKeys.isEmpty()) {
            return;
        }
        filterLocationKeys(locationKeys)
                .parallelStream()
                .forEach(lk -> columnSourceManager.addLocation(locationProvider.getTableLocation(lk.get())));
    }

Note the parallelStream() call.

Now, I expect that we're reading Parquet file metadata under RegionedColumnSourceManager.initialize(), which delegates to ``RegionedColumnSourceManager.update(true), which uses a parallel stream when checking empty entries for existence (see the true` last argument to `StreamSupport.stream`):

        final Collection<EmptyTableLocationEntry> entriesToInclude = StreamSupport.stream(Spliterators.spliterator(
                                emptyTableLocations.iterator(),
                                emptyTableLocations.size(),
                                Spliterator.IMMUTABLE | Spliterator.DISTINCT | Spliterator.NONNULL),
                        true) //@formatter:on
                .peek(EmptyTableLocationEntry::refresh)
                .filter((final EmptyTableLocationEntry emptyEntry) -> {

The first call to EmptyTableLocationEntry.refresh is going to cause the actual TableLocation.refresh/TableLocation.subscribe call, which will force initialization.

[malhotrashivam]

From what I can see in my test, EmptyTableLocationEntry.refresh calls ParquetTableLocation.refresh which is an empty method. The actual reading of parquet metadata happens in

final RowSet locationRowSet = emptyEntry.location.getRowSet();

which calls AbstractTableLocation::getRowSet and then ParquetTableLocation::initialize

[rcaudy]

From what I can see in my test, EmptyTableLocationEntry.refresh calls ParquetTableLocation.refresh which is an empty method. The actual reading of parquet metadata happens in

final RowSet locationRowSet = emptyEntry.location.getRowSet();

which calls AbstractTableLocation::getRowSet and then ParquetTableLocation::initialize

And the getRowSet() is also on a parallel stream, so we should be good here.
Since you said you're testing agreed, let's declare victory.