andygrove commented on issue #7451: URL: https://github.com/apache/datafusion-comet/issues/7451#issuecomment-6595819216
I looked into the Hadoop side of this. The likely cause of the Spark 4.1.3 vs 4.2.0 gap is narrower than "S3A in 3.5.0 is faster": **Hadoop 3.5.0 changed the default S3A input stream from the classic `S3AInputStream` to the Analytics Accelerator Library (AAL).** This is HADOOP-19559 (apache/hadoop commit `9c44fa23f4`, a one-line change): ``` hadoop-aws/.../impl/streams/StreamIntegration.java:80 - DEFAULT_STREAM_TYPE = InputStreamType.Classic; // 3.4.2 + DEFAULT_STREAM_TYPE = InputStreamType.Analytics; // 3.5.0 ``` Hadoop 3.4.2 already ships AAL (1.2.1) but only uses it with `fs.s3a.input.stream.type=analytics`. The 3.5.0 `reading.md` still says the default is `classic`, but the code says otherwise. AAL 1.3.1 defaults that line up with what we measured: | AAL default | Effect | |---|---| | Small-object prefetch, threshold 8 MB | A few-KB file (`store`, `call_center`, ...) is read with one GET. The classic stream issues a GET for the footer and another when it seeks back for the data. | | Footer prefetch (32 KB, 1 MB for large files) and a parsed Parquet footer store (45 entries) | The footer and data reads collapse into one request. | | Single shared block cache across streams (1 s data TTL) | Spark opens each file twice (footer reader, then record reader). The second open is served from memory. | | Object metadata cache (24 h TTL, 5,000 entries) | No repeated HEADs. | | Parquet-aware row-group prefetch | Column chunks are fetched ahead, in parallel. | That fits the symptoms: identical plans, a broad slowdown that hurts short queries most, and 3-5x slower reads of one-file dimension tables on 4.1.3. All of it points at per-file request round trips rather than decoding. On the Spark side, nothing between v4.1.3 and v4.2.0 changes Parquet I/O. SPARK-55273 (`readFooter()` instead of `open().getFooter()`) only saves CPU, not reads. The Parquet 1.16 to 1.17 bump and SPARK-55885 (boolean decoding) may help slightly but do not explain per-file latency. Spark does not set `fs.s3a.input.stream.type`, so 4.2.0 just inherits Hadoop's new default. Comet's native scan reads through `object_store` and never uses the S3A input stream, so the switch speeds up the Spark baseline but not Comet's scan. Comet's own time still drops about 10-14% on 4.2.0; that may come from paths that still go through Hadoop (listing, fallback scans), but I have not checked. A minor note on versions: Spark 4.1.3's pom pins AAL 1.3.0, but Hadoop 3.4.2 pins 1.2.1, which is probably what the image contains. It does not matter here because the classic stream is used on 4.1.3 either way. ### Confirming it This is simpler than swapping Hadoop jars into 4.1.3. Either run isolates the stream type: - Spark 4.2.0 with `spark.hadoop.fs.s3a.input.stream.type=classic`, expected to move toward the 4.1.3 baseline (~1,336 s). - Spark 4.1.3 with `spark.hadoop.fs.s3a.input.stream.type=analytics`, expected to move toward the 4.2.0 baseline (~994 s). If confirmed, the follow-up for Comet is to bring the same behavior to the native `object_store` scan: a single GET for small files, fetching the footer together with the data, and reusing fetched bytes across the footer read and the data read of the same file. -- 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]
