shunping commented on code in PR #40142:
URL: https://github.com/apache/beam/pull/40142#discussion_r4041800636
##########
sdks/java/io/parquet/src/main/java/org/apache/beam/sdk/io/parquet/ParquetIO.java:
##########
@@ -736,6 +746,34 @@ private ParquetFileReader
getParquetFileReader(ReadableFile file) throws Excepti
return ParquetFileReader.open(new
BeamParquetInputFile(file.openSeekable()), options);
}
+ /**
+ * Returns the row group metadata of {@code file}, reading the Parquet
footer only if it is
+ * not already cached.
+ *
+ * <p>{@link #getInitialRestriction}, {@link #split}, {@link
#newTracker} and {@link #getSize}
+ * all need nothing from the file but its row group metadata, and the
runner may invoke them
+ * repeatedly for the same element: {@link #getSize} and {@link
#newTracker} are called on
+ * every dynamic split attempt. Opening a {@link ParquetFileReader} each
time re-reads the
+ * footer, which costs a round trip per call on object stores such as
GCS or S3. Caching the
+ * last file's row groups reduces that to one footer read per file.
+ *
+ * <p>This is synchronized because a runner may invoke the size and
tracker callbacks from a
+ * different thread than the one processing the bundle, concurrently
with {@link
+ * #processElement}.
+ */
+ private synchronized List<BlockMetaData> getRowGroups(ReadableFile file)
throws Exception {
Review Comment:
Yes, it is an optimization.
Previously we retrieve the row group information for every getSize() and
newTracker() on every dynamic split attempt or progress probing. In one of the
tests, this fix reduced the number of metadata queries by ~80%.
--
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]
