You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "Baunsgaard (via GitHub)" <gi...@apache.org> on 2026/06/09 08:30:00 UTC
[PR] Spark tests cache rewrite input [iceberg]
Baunsgaard opened a new pull request, #16740:
URL: https://github.com/apache/iceberg/pull/16740
## What
`TestRewriteDataFilesAction` is the slowest single class in the Spark core test
module. Each test materializes a large (`SCALE = 400000`-row) input table via a
Spark write before exercising the rewrite under test, and many tests reuse the
same input shape across the `formatVersion` matrix.
This caches the written input data files keyed by table shape (`formatVersion`,
spec, `files`, `rows`, `partitions`, properties) and reuses them by re-appending
the cached `DataFile`s to a fresh table. The expensive Spark write of the input
now runs once per JVM fork instead of once per test; the rewrite under test still
runs per test on its own fresh table.
Applied identically to Spark 3.5, 4.0 and 4.1.
## Why it is safe
- The generated data is deterministic (fixed `Random(42)` seed), so reuse is
byte-identical to regenerating it.
- Cached files live in a static `@TempDir`, so they survive across tests (not wiped
by the per-test temp dir) and are cleaned up after the class.
- `includeColumnStats()` is used when collecting the cached files so lower/upper
bounds and value counts are preserved on re-append.
- The rewrite under test is unchanged and still runs per test, so no assertion is
weakened.
## Results (local, JDK 17, 32 cores)
| Scope | baseline | with cache |
|---|---:|---:|
| `TestRewriteDataFilesAction` (single-thread) | 705s | 455s |
| full `iceberg-spark-3.5_2.13` core module (`testParallelism=auto`) | 18m56s | 14m57s |
Test/skip counts are unchanged at both class and module level:
Spark 3.5 = 168 tests / 6 skipped / 0 failed; Spark 4.0 & 4.1 = 171 / 6 / 0.
## Notes
- Scoped to the `createTable(int)` / `createTablePartitioned(...)` helpers; the few
in-test `writeRecords(..., SCALE, ...)` call sites are not yet cached.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on code in PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#discussion_r3389952253
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -143,6 +150,15 @@ public class TestRewriteDataFilesAction extends TestBase {
@TempDir private File tableDir;
private static final int SCALE = 400000;
+ // Cache of pre-written input data files keyed by table shape (schema/spec/props are
+ // fixed per key), so identical large inputs are materialized via Spark only once per JVM
+ // fork and reused by every test that asks for the same shape. The Spark write of SCALE
+ // rows dominates these tests; the rewrite under test still runs per test on a fresh table.
+ @TempDir private static Path inputCacheDir;
+ private static final Map<String, List<DataFile>> INPUT_FILE_CACHE = Maps.newConcurrentMap();
Review Comment:
Okay, accordingly added an `@AfterAll` that clears the cache + lock map and resets the seq. `@AfterAll` runs once after all tests, so within-run cross-test caching is unchanged; it only stops a second in-JVM run (IDE re-run) from returning DataFiles pointing into the recreated `@TempDir`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka merged PR #16740:
URL: https://github.com/apache/iceberg/pull/16740
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#issuecomment-4751400621
Sure, I took a look at your suggestion and it makes sense. Up to you which version you like best.
The pr is open to contributor modification so you can edit directly if you wish.
Otherwise I can change the commit to yours next week and set you as coauthor.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on code in PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#discussion_r3389966336
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2305,23 +2321,95 @@ protected Table createTable() {
* @return the created table
*/
protected Table createTable(int files) {
+ String key = String.format("unpartitioned|fv=%d|files=%d|rows=%d", formatVersion, files, SCALE);
+ List<DataFile> inputFiles =
+ cachedInputFiles(
+ key,
+ () -> {
+ Table golden = createTable();
+ writeRecords(files, SCALE);
+ golden.refresh();
+ return golden;
+ });
Table table = createTable();
- writeRecords(files, SCALE);
+ appendInputFiles(table, inputFiles);
table.refresh();
return table;
}
protected Table createTablePartitioned(
int partitions, int files, int numRecords, Map<String, String> options) {
PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("c1").truncate("c2", 2).build();
+ String key =
+ String.format(
+ "partitioned|fv=%d|spec=%s|opts=%s|files=%d|rows=%d|partitions=%d",
+ formatVersion, spec, new TreeMap<>(options), files, numRecords, partitions);
+ List<DataFile> inputFiles =
+ cachedInputFiles(
+ key,
+ () -> {
+ Table golden = TABLES.create(SCHEMA, spec, options, tableLocation);
+ assertThat(golden.currentSnapshot()).as("Table must be empty").isNull();
+ writeRecords(files, numRecords, partitions);
+ golden.refresh();
+ return golden;
+ });
Table table = TABLES.create(SCHEMA, spec, options, tableLocation);
assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
-
- writeRecords(files, numRecords, partitions);
+ appendInputFiles(table, inputFiles);
table.refresh();
return table;
}
+ /**
+ * Returns the input data files for a given table shape, materializing them with Spark exactly
+ * once per JVM fork and reusing them afterwards. On a cache miss the {@code goldenBuilder} writes
+ * the data into a stable cache location (kept alive for the whole class via a static {@link
+ * TempDir}); on a hit the cached {@link DataFile}s are returned and re-appended to a fresh table
+ * by {@link #appendInputFiles}. The data is deterministic (fixed RNG seed) so reuse is
+ * byte-identical to regenerating it.
+ */
+ private List<DataFile> cachedInputFiles(String key, Supplier<Table> goldenBuilder) {
+ List<DataFile> cached = INPUT_FILE_CACHE.get(key);
+ if (cached != null) {
+ return cached;
+ }
+ // Serialize builds per key: concurrent callers requesting the same table shape block on the
+ // first build and then reuse its result, instead of materializing identical input twice. The
+ // heavy Spark write happens outside any map lock, so distinct shapes can still build in
+ // parallel.
+ Object lock = INPUT_CACHE_LOCKS.computeIfAbsent(key, ignored -> new Object());
+ synchronized (lock) {
+ List<DataFile> existing = INPUT_FILE_CACHE.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ String savedLocation = this.tableLocation;
+ try {
+ this.tableLocation =
+ inputCacheDir.resolve("input-" + INPUT_CACHE_SEQ.incrementAndGet()).toUri().toString();
Review Comment:
Since the cache explicitly saves the files in a different folder, and actively copies them over for the individual tests it does not invalidate any of the `shouldHaveNoOrphans` calls, but to be defensive, sure we can add it.
Therefore, accordingly I a comment at `shouldHaveNoOrphans` noting cached inputs live under the static `inputCacheDir`, outside `table.location()`, so `deleteOrphanFiles` intentionally doesn't scan them.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#issuecomment-4672416387
> Nice, this is right improvement. `SCALE` write dominates these tests, and caching the golden files per table shape is exactly where the ~8% win should come from. Re-appending the same physical files is spec-legal too, since sequence numbers and `firstRowId` are assigned on each append.
>
> ...
>
> Once those are addressed, happy to do one more pass.
Thanks for the review @laskoviymishka ! I have addressed all the four blockers!
One Nit: I left the DCL-vs-`PER_METHOD` and virtual-dispatch nits as-is for now. I know my implementation on the locking is hypothetical for future parallel execution and the tests are run single threaded now, but i thought it would be fine to make it thread-safe already now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#issuecomment-4752512101
Thanks for taking a look @Baunsgaard! I can open a PR with the edits. I mainly want this part of the tests to be more maintainable, the coauthor part is not a concern 😄
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on code in PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#discussion_r3389946136
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2305,23 +2321,95 @@ protected Table createTable() {
* @return the created table
*/
protected Table createTable(int files) {
+ String key = String.format("unpartitioned|fv=%d|files=%d|rows=%d", formatVersion, files, SCALE);
Review Comment:
I think this is very speculative about future commits, but fine. I pulled the row-group size into a constant (`INPUT_PARQUET_ROW_GROUP_SIZE_BYTES`) and folded it into the key (`rowGroup=%d`), mirroring the partitioned path. A future `createTable()` property override now changes the key instead of silently colliding.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#issuecomment-4731241395
@kevinjqliu Could you take a look, thanks!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on code in PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#discussion_r3389938983
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2305,23 +2321,95 @@ protected Table createTable() {
* @return the created table
*/
protected Table createTable(int files) {
+ String key = String.format("unpartitioned|fv=%d|files=%d|rows=%d", formatVersion, files, SCALE);
+ List<DataFile> inputFiles =
+ cachedInputFiles(
+ key,
+ () -> {
+ Table golden = createTable();
+ writeRecords(files, SCALE);
+ golden.refresh();
+ return golden;
+ });
Table table = createTable();
- writeRecords(files, SCALE);
+ appendInputFiles(table, inputFiles);
table.refresh();
return table;
}
protected Table createTablePartitioned(
int partitions, int files, int numRecords, Map<String, String> options) {
PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("c1").truncate("c2", 2).build();
+ String key =
+ String.format(
+ "partitioned|fv=%d|spec=%s|opts=%s|files=%d|rows=%d|partitions=%d",
+ formatVersion, spec, new TreeMap<>(options), files, numRecords, partitions);
+ List<DataFile> inputFiles =
+ cachedInputFiles(
+ key,
+ () -> {
+ Table golden = TABLES.create(SCHEMA, spec, options, tableLocation);
+ assertThat(golden.currentSnapshot()).as("Table must be empty").isNull();
+ writeRecords(files, numRecords, partitions);
+ golden.refresh();
+ return golden;
+ });
Table table = TABLES.create(SCHEMA, spec, options, tableLocation);
assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
-
- writeRecords(files, numRecords, partitions);
+ appendInputFiles(table, inputFiles);
table.refresh();
return table;
}
+ /**
+ * Returns the input data files for a given table shape, materializing them with Spark exactly
+ * once per JVM fork and reusing them afterwards. On a cache miss the {@code goldenBuilder} writes
+ * the data into a stable cache location (kept alive for the whole class via a static {@link
+ * TempDir}); on a hit the cached {@link DataFile}s are returned and re-appended to a fresh table
+ * by {@link #appendInputFiles}. The data is deterministic (fixed RNG seed) so reuse is
+ * byte-identical to regenerating it.
+ */
+ private List<DataFile> cachedInputFiles(String key, Supplier<Table> goldenBuilder) {
+ List<DataFile> cached = INPUT_FILE_CACHE.get(key);
+ if (cached != null) {
+ return cached;
+ }
+ // Serialize builds per key: concurrent callers requesting the same table shape block on the
+ // first build and then reuse its result, instead of materializing identical input twice. The
+ // heavy Spark write happens outside any map lock, so distinct shapes can still build in
+ // parallel.
+ Object lock = INPUT_CACHE_LOCKS.computeIfAbsent(key, ignored -> new Object());
+ synchronized (lock) {
+ List<DataFile> existing = INPUT_FILE_CACHE.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ String savedLocation = this.tableLocation;
+ try {
+ this.tableLocation =
+ inputCacheDir.resolve("input-" + INPUT_CACHE_SEQ.incrementAndGet()).toUri().toString();
+ Table golden = goldenBuilder.get();
+ // includeColumnStats() is required: a plain scan drops lower/upper bounds and
+ // value counts, and re-appending stat-less files breaks tests that read bounds.
+ List<DataFile> built =
+ Streams.stream(golden.newScan().includeColumnStats().planFiles())
Review Comment:
Okay! Wrapped the scan in try-with-resources and collect inside, mapping `IOException` to `UncheckedIOException`. No more reader leak per cache miss.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
laskoviymishka commented on code in PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#discussion_r3389616665
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2305,23 +2321,95 @@ protected Table createTable() {
* @return the created table
*/
protected Table createTable(int files) {
+ String key = String.format("unpartitioned|fv=%d|files=%d|rows=%d", formatVersion, files, SCALE);
+ List<DataFile> inputFiles =
+ cachedInputFiles(
+ key,
+ () -> {
+ Table golden = createTable();
+ writeRecords(files, SCALE);
+ golden.refresh();
+ return golden;
+ });
Table table = createTable();
- writeRecords(files, SCALE);
+ appendInputFiles(table, inputFiles);
table.refresh();
return table;
}
protected Table createTablePartitioned(
int partitions, int files, int numRecords, Map<String, String> options) {
PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("c1").truncate("c2", 2).build();
+ String key =
+ String.format(
+ "partitioned|fv=%d|spec=%s|opts=%s|files=%d|rows=%d|partitions=%d",
+ formatVersion, spec, new TreeMap<>(options), files, numRecords, partitions);
+ List<DataFile> inputFiles =
+ cachedInputFiles(
+ key,
+ () -> {
+ Table golden = TABLES.create(SCHEMA, spec, options, tableLocation);
+ assertThat(golden.currentSnapshot()).as("Table must be empty").isNull();
+ writeRecords(files, numRecords, partitions);
+ golden.refresh();
+ return golden;
+ });
Table table = TABLES.create(SCHEMA, spec, options, tableLocation);
assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
-
- writeRecords(files, numRecords, partitions);
+ appendInputFiles(table, inputFiles);
table.refresh();
return table;
}
+ /**
+ * Returns the input data files for a given table shape, materializing them with Spark exactly
+ * once per JVM fork and reusing them afterwards. On a cache miss the {@code goldenBuilder} writes
+ * the data into a stable cache location (kept alive for the whole class via a static {@link
+ * TempDir}); on a hit the cached {@link DataFile}s are returned and re-appended to a fresh table
+ * by {@link #appendInputFiles}. The data is deterministic (fixed RNG seed) so reuse is
+ * byte-identical to regenerating it.
+ */
+ private List<DataFile> cachedInputFiles(String key, Supplier<Table> goldenBuilder) {
+ List<DataFile> cached = INPUT_FILE_CACHE.get(key);
+ if (cached != null) {
+ return cached;
+ }
+ // Serialize builds per key: concurrent callers requesting the same table shape block on the
+ // first build and then reuse its result, instead of materializing identical input twice. The
+ // heavy Spark write happens outside any map lock, so distinct shapes can still build in
+ // parallel.
+ Object lock = INPUT_CACHE_LOCKS.computeIfAbsent(key, ignored -> new Object());
+ synchronized (lock) {
+ List<DataFile> existing = INPUT_FILE_CACHE.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ String savedLocation = this.tableLocation;
+ try {
+ this.tableLocation =
+ inputCacheDir.resolve("input-" + INPUT_CACHE_SEQ.incrementAndGet()).toUri().toString();
Review Comment:
Caching the input files outside the per-test table location quietly weakens the orphan-file coverage. `shouldHaveNoOrphans(table)` runs `deleteOrphanFiles`, which by default only scans `table.location()` (the per-test `tableDir`), but the cached inputs live under this static `inputCacheDir`, outside that prefix — so the `startsWith(location)` filter in `DeleteOrphanFilesSparkAction` never sees them.
The tests still pass, but if a bug ever leaked a cached input path into a post-rewrite snapshot, the no-orphans assertions wouldn't catch it — we've lost diagnostic value, not gained a false failure.
I'd at minimum drop a comment near `shouldHaveNoOrphans` noting the cached inputs are out of scan scope by design; or, if we want to keep the invariant strong, add a manifest-scan assertion that the fresh table's current snapshot contains no `inputCacheDir`-prefixed paths. Same in v4.0 and v4.1.
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2305,23 +2321,95 @@ protected Table createTable() {
* @return the created table
*/
protected Table createTable(int files) {
+ String key = String.format("unpartitioned|fv=%d|files=%d|rows=%d", formatVersion, files, SCALE);
Review Comment:
The unpartitioned key drops the table-properties dimension that the partitioned path includes. The partitioned key folds in `opts` via `new TreeMap<>(options)`, but here the key is just `unpartitioned|fv|files|rows` while `createTable()` sets `PARQUET_ROW_GROUP_SIZE_BYTES=20*1024`, which shapes `split_offsets` on every cached DataFile.
It's dormant today because the value is hardcoded, but it's a silent shape collision waiting to happen: if any subclass or future test overrides `createTable()`'s properties, the cache serves files with mismatched split offsets, and tests like `testBinPackSplitLargeFile` that lean on accurate offsets fail in a way that's very hard to trace back to here.
I'd mirror the partitioned path and fold the relevant properties into the unpartitioned key. wdyt? Same in v4.0 and v4.1.
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -143,6 +150,15 @@ public class TestRewriteDataFilesAction extends TestBase {
@TempDir private File tableDir;
private static final int SCALE = 400000;
+ // Cache of pre-written input data files keyed by table shape (schema/spec/props are
+ // fixed per key), so identical large inputs are materialized via Spark only once per JVM
+ // fork and reused by every test that asks for the same shape. The Spark write of SCALE
+ // rows dominates these tests; the rewrite under test still runs per test on a fresh table.
+ @TempDir private static Path inputCacheDir;
+ private static final Map<String, List<DataFile>> INPUT_FILE_CACHE = Maps.newConcurrentMap();
Review Comment:
I think there's a subtle lifecycle issue when the class runs twice in one JVM. `INPUT_FILE_CACHE` is static and never cleared, but `inputCacheDir` is a static `@TempDir`, so JUnit deletes and recreates it on a second run in the same JVM (IDE re-run, `forkCount=0`, suite aggregation). On that second run the cache still holds DataFiles whose `location()` points into the first, now-deleted directory, so a cache hit hands back dangling paths and you get a `FileNotFoundException` deep in the Spark Parquet reader.
Invisible in normal Gradle CI since each fork is a fresh JVM, but it bites IntelliJ re-runs.
Either clear the maps and reset the seq in `@AfterAll`, or key the cache by `inputCacheDir.toAbsolutePath()` so stale entries can't match a new directory. Either is fine, but one of them should be in. Same in v4.0 and v4.1.
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java:
##########
@@ -2305,23 +2321,95 @@ protected Table createTable() {
* @return the created table
*/
protected Table createTable(int files) {
+ String key = String.format("unpartitioned|fv=%d|files=%d|rows=%d", formatVersion, files, SCALE);
+ List<DataFile> inputFiles =
+ cachedInputFiles(
+ key,
+ () -> {
+ Table golden = createTable();
+ writeRecords(files, SCALE);
+ golden.refresh();
+ return golden;
+ });
Table table = createTable();
- writeRecords(files, SCALE);
+ appendInputFiles(table, inputFiles);
table.refresh();
return table;
}
protected Table createTablePartitioned(
int partitions, int files, int numRecords, Map<String, String> options) {
PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("c1").truncate("c2", 2).build();
+ String key =
+ String.format(
+ "partitioned|fv=%d|spec=%s|opts=%s|files=%d|rows=%d|partitions=%d",
+ formatVersion, spec, new TreeMap<>(options), files, numRecords, partitions);
+ List<DataFile> inputFiles =
+ cachedInputFiles(
+ key,
+ () -> {
+ Table golden = TABLES.create(SCHEMA, spec, options, tableLocation);
+ assertThat(golden.currentSnapshot()).as("Table must be empty").isNull();
+ writeRecords(files, numRecords, partitions);
+ golden.refresh();
+ return golden;
+ });
Table table = TABLES.create(SCHEMA, spec, options, tableLocation);
assertThat(table.currentSnapshot()).as("Table must be empty").isNull();
-
- writeRecords(files, numRecords, partitions);
+ appendInputFiles(table, inputFiles);
table.refresh();
return table;
}
+ /**
+ * Returns the input data files for a given table shape, materializing them with Spark exactly
+ * once per JVM fork and reusing them afterwards. On a cache miss the {@code goldenBuilder} writes
+ * the data into a stable cache location (kept alive for the whole class via a static {@link
+ * TempDir}); on a hit the cached {@link DataFile}s are returned and re-appended to a fresh table
+ * by {@link #appendInputFiles}. The data is deterministic (fixed RNG seed) so reuse is
+ * byte-identical to regenerating it.
+ */
+ private List<DataFile> cachedInputFiles(String key, Supplier<Table> goldenBuilder) {
+ List<DataFile> cached = INPUT_FILE_CACHE.get(key);
+ if (cached != null) {
+ return cached;
+ }
+ // Serialize builds per key: concurrent callers requesting the same table shape block on the
+ // first build and then reuse its result, instead of materializing identical input twice. The
+ // heavy Spark write happens outside any map lock, so distinct shapes can still build in
+ // parallel.
+ Object lock = INPUT_CACHE_LOCKS.computeIfAbsent(key, ignored -> new Object());
+ synchronized (lock) {
+ List<DataFile> existing = INPUT_FILE_CACHE.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ String savedLocation = this.tableLocation;
+ try {
+ this.tableLocation =
+ inputCacheDir.resolve("input-" + INPUT_CACHE_SEQ.incrementAndGet()).toUri().toString();
+ Table golden = goldenBuilder.get();
+ // includeColumnStats() is required: a plain scan drops lower/upper bounds and
+ // value counts, and re-appending stat-less files breaks tests that read bounds.
+ List<DataFile> built =
+ Streams.stream(golden.newScan().includeColumnStats().planFiles())
Review Comment:
`planFiles()` returns a `CloseableIterable`, and handing it to `Streams.stream(Iterable)` means it never gets closed — so every cache miss leaks the manifest readers it holds open. One leak per distinct key per fork; on a long-lived JVM that's FD pressure, on Windows it leaves the manifest files locked, and `@TempDir` cleanup can warn at teardown.
I'd wrap the scan in try-with-resources and collect inside the try:
```java
List<DataFile> built;
try (CloseableIterable<FileScanTask> tasks = golden.newScan().includeColumnStats().planFiles()) {
built =
Streams.stream(tasks)
.map(FileScanTask::file)
.map(DataFile::copy)
.collect(ImmutableList.toImmutableList());
}
INPUT_FILE_CACHE.put(key, built);
return built;
```
(The pre-existing `planFiles()` calls elsewhere in the file have the same shape but are out of scope here.) Same in v4.0 and v4.1.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "Baunsgaard (via GitHub)" <gi...@apache.org>.
Baunsgaard commented on PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#issuecomment-4669576269
Small update on the times:
Core-job wall time, **before** = baseline main run, **after** = mean of 3 green
PR runs (per Spark version, averaged across the JVM x Scala matrix to smooth
runner noise):
| Spark core jobs | Before | After (mean of 3 runs) | Δ |
|---|---:|---:|---:|
| 3.5 | 40.6 min | 37.5 min | −7.6% |
| 4.0 | 47.3 min | 42.9 min | −9.3% |
| 4.1 | 42.9 min | 39.1 min | −8.9% |
| **Average** | **42.8 min** | **39.2 min** | **−8.4%** |
The per-run overall core average is stable across the three runs (39.4 / 39.3 /
38.9 min), so the ~8% improvement is consistent rather than a single lucky run.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#issuecomment-4733245102
Thanks for the improvement @Baunsgaard, the changes looks good to me.
I did have one nit about making the changes easier to read. I think we can get rid of the locks and atomic integer. And possibly isolate the changes to the `writeRecords` function.
I had codex generate something similar if you would like to take a look, https://github.com/apache/iceberg/compare/main...kevinjqliu:iceberg:kevinjqliu/codex-cache-rewrite-data-files-input?expand=1
Again, its mostly for readability and maintainability -- the effect is the same 😄
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark: Spark tests cache rewrite input [iceberg]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on PR #16740:
URL: https://github.com/apache/iceberg/pull/16740#issuecomment-4754862792
opened https://github.com/apache/iceberg/pull/16882 let me know what you think!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org