From 54863ec63ef5c82b1669f28a8195f786a1491ecf Mon Sep 17 00:00:00 2001 From: Erik Holmer <49652522+eholmer-pltr@users.noreply.github.com> Date: Tue, 16 Jun 2026 12:03:36 +0200 Subject: [PATCH] Core: Avoid HEAD on metadata file in synthetic metadata-table tasks HistoryTable, SnapshotsTable, MetadataLogEntriesTable, RefsTable, ManifestsTable, AllManifestsTable, and PartitionsTable build a StaticDataTask whose rows come from in-memory TableMetadata (or from pre-walked manifest data). They wrap a DataFile around metadataFileLocation purely so FileScanTask.file() has a path to return; the bytes are never read. The old StaticDataTask.of(InputFile, ...) entry point built that DataFile via DataFiles.Builder.withInputFile, which calls InputFile.getLength() to populate fileSizeInBytes -- triggering a HEAD/GetObject against the metadata.json on object-store FileIO. That size is never consulted by the DataTask read path in any engine (Spark's RowDataReader.open and Flink's DataTaskReader.open both gate file-size-aware logic on !task.isDataTask()). This replaces StaticDataTask.of(InputFile, ...) with StaticDataTask.of(String location, ...), which builds the DataFile via withPath + withFileSizeInBytes(0L), and migrates the seven metadata tables plus TestDataTaskParser to it. The old InputFile overload and its private constructor (the only remaining getLength call site) are removed, so no construction path performs I/O against the synthetic location. task.length() for these tasks now returns 0 instead of the metadata.json file size. Precedent: AllManifestsTable.ManifestListReadTask.length() already returns a hard-coded 8192 with the comment "return a generic length to avoid looking up the actual length". --- .../org/apache/iceberg/AllManifestsTable.java | 3 +- .../java/org/apache/iceberg/HistoryTable.java | 2 +- .../org/apache/iceberg/ManifestsTable.java | 3 +- .../iceberg/MetadataLogEntriesTable.java | 2 +- .../org/apache/iceberg/PartitionsTable.java | 4 +- .../java/org/apache/iceberg/RefsTable.java | 2 +- .../org/apache/iceberg/SnapshotsTable.java | 2 +- .../org/apache/iceberg/StaticDataTask.java | 36 +++--- .../apache/iceberg/TestDataTaskParser.java | 2 +- .../apache/iceberg/TestStaticDataTask.java | 104 ++++++++++++++++++ 10 files changed, 129 insertions(+), 31 deletions(-) create mode 100644 core/src/test/java/org/apache/iceberg/TestStaticDataTask.java diff --git a/core/src/main/java/org/apache/iceberg/AllManifestsTable.java b/core/src/main/java/org/apache/iceberg/AllManifestsTable.java index 2435de62f0f9..beb27283ec59 100644 --- a/core/src/main/java/org/apache/iceberg/AllManifestsTable.java +++ b/core/src/main/java/org/apache/iceberg/AllManifestsTable.java @@ -145,8 +145,7 @@ protected CloseableIterable doPlanFiles() { snap.snapshotId()); } else { return StaticDataTask.of( - io.newInputFile( - ((BaseTable) table()).operations().current().metadataFileLocation()), + ((BaseTable) table()).operations().current().metadataFileLocation(), MANIFEST_FILE_SCHEMA, schema(), snap.allManifests(io), diff --git a/core/src/main/java/org/apache/iceberg/HistoryTable.java b/core/src/main/java/org/apache/iceberg/HistoryTable.java index c2bd01ea4a8e..ee466f6b50c6 100644 --- a/core/src/main/java/org/apache/iceberg/HistoryTable.java +++ b/core/src/main/java/org/apache/iceberg/HistoryTable.java @@ -66,7 +66,7 @@ MetadataTableType metadataTableType() { private DataTask task(TableScan scan) { return StaticDataTask.of( - table().io().newInputFile(table().operations().current().metadataFileLocation()), + table().operations().current().metadataFileLocation(), schema(), scan.schema(), table().history(), diff --git a/core/src/main/java/org/apache/iceberg/ManifestsTable.java b/core/src/main/java/org/apache/iceberg/ManifestsTable.java index 5c850ec6f9fb..24c5ef273ec8 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestsTable.java +++ b/core/src/main/java/org/apache/iceberg/ManifestsTable.java @@ -81,8 +81,7 @@ protected DataTask task(TableScan scan) { Map specs = Maps.newHashMap(table().specs()); return StaticDataTask.of( - io.newInputFile( - location != null ? location : table().operations().current().metadataFileLocation()), + location != null ? location : table().operations().current().metadataFileLocation(), schema(), scan.schema(), scan.snapshot().allManifests(io), diff --git a/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java b/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java index 3cffee37dc50..79dcb4c2a679 100644 --- a/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java +++ b/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java @@ -65,7 +65,7 @@ private DataTask task(TableScan scan) { new TableMetadata.MetadataLogEntry( current.lastUpdatedMillis(), current.metadataFileLocation())); return StaticDataTask.of( - table().io().newInputFile(current.metadataFileLocation()), + current.metadataFileLocation(), schema(), scan.schema(), metadataLogEntries, diff --git a/core/src/main/java/org/apache/iceberg/PartitionsTable.java b/core/src/main/java/org/apache/iceberg/PartitionsTable.java index 10366db5a55d..bde4838b8f6c 100644 --- a/core/src/main/java/org/apache/iceberg/PartitionsTable.java +++ b/core/src/main/java/org/apache/iceberg/PartitionsTable.java @@ -153,7 +153,7 @@ private DataTask task(StaticTableScan scan) { if (unpartitionedTable) { // the table is unpartitioned, partitions contains only the root partition return StaticDataTask.of( - io().newInputFile(table().operations().current().metadataFileLocation()), + table().operations().current().metadataFileLocation(), schema(), scan.schema(), partitions, @@ -170,7 +170,7 @@ private DataTask task(StaticTableScan scan) { root.lastUpdatedSnapshotId)); } else { return StaticDataTask.of( - io().newInputFile(table().operations().current().metadataFileLocation()), + table().operations().current().metadataFileLocation(), schema(), scan.schema(), partitions, diff --git a/core/src/main/java/org/apache/iceberg/RefsTable.java b/core/src/main/java/org/apache/iceberg/RefsTable.java index f906109bd531..ab8911694999 100644 --- a/core/src/main/java/org/apache/iceberg/RefsTable.java +++ b/core/src/main/java/org/apache/iceberg/RefsTable.java @@ -60,7 +60,7 @@ public Schema schema() { private DataTask task(BaseTableScan scan) { Collection refNames = table().refs().keySet(); return StaticDataTask.of( - table().io().newInputFile(table().operations().current().metadataFileLocation()), + table().operations().current().metadataFileLocation(), schema(), scan.schema(), refNames, diff --git a/core/src/main/java/org/apache/iceberg/SnapshotsTable.java b/core/src/main/java/org/apache/iceberg/SnapshotsTable.java index f948c5578345..d031be74faa7 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotsTable.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotsTable.java @@ -59,7 +59,7 @@ public Schema schema() { private DataTask task(BaseTableScan scan) { return StaticDataTask.of( - table().io().newInputFile(table().operations().current().metadataFileLocation()), + table().operations().current().metadataFileLocation(), schema(), scan.schema(), table().snapshots(), diff --git a/core/src/main/java/org/apache/iceberg/StaticDataTask.java b/core/src/main/java/org/apache/iceberg/StaticDataTask.java index 1a396f0bfc7e..ababe037e129 100644 --- a/core/src/main/java/org/apache/iceberg/StaticDataTask.java +++ b/core/src/main/java/org/apache/iceberg/StaticDataTask.java @@ -25,7 +25,6 @@ import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.io.CloseableIterable; -import org.apache.iceberg.io.InputFile; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Lists; @@ -33,17 +32,27 @@ class StaticDataTask implements DataTask { + /** + * Builds a {@link StaticDataTask} from in-memory rows. The {@code metadataLocation} is kept on + * the synthetic {@link DataFile} for identification only and is never read, so its size is + * reported as 0 rather than looked up. + */ static DataTask of( - InputFile metadata, + String metadataLocation, Schema tableSchema, Schema projectedSchema, Iterable values, Function transform) { - return new StaticDataTask( - metadata, - tableSchema, - projectedSchema, - Lists.newArrayList(Iterables.transform(values, transform::apply)).toArray(new Row[0])); + Row[] rows = + Lists.newArrayList(Iterables.transform(values, transform::apply)).toArray(new Row[0]); + DataFile syntheticFile = + DataFiles.builder(PartitionSpec.unpartitioned()) + .withPath(metadataLocation) + .withFileSizeInBytes(0L) + .withRecordCount(rows.length) + .withFormat(FileFormat.METADATA) + .build(); + return new StaticDataTask(syntheticFile, tableSchema, projectedSchema, rows); } private final DataFile metadataFile; @@ -51,19 +60,6 @@ static DataTask of( private final Schema tableSchema; private final Schema projectedSchema; - private StaticDataTask( - InputFile metadata, Schema tableSchema, Schema projectedSchema, StructLike[] rows) { - this.tableSchema = tableSchema; - this.projectedSchema = projectedSchema; - this.metadataFile = - DataFiles.builder(PartitionSpec.unpartitioned()) - .withInputFile(metadata) - .withRecordCount(rows.length) - .withFormat(FileFormat.METADATA) - .build(); - this.rows = rows; - } - StaticDataTask( DataFile metadataFile, Schema tableSchema, Schema projectedSchema, StructLike[] rows) { this.tableSchema = tableSchema; diff --git a/core/src/test/java/org/apache/iceberg/TestDataTaskParser.java b/core/src/test/java/org/apache/iceberg/TestDataTaskParser.java index 03065abe8744..826501893340 100644 --- a/core/src/test/java/org/apache/iceberg/TestDataTaskParser.java +++ b/core/src/test/java/org/apache/iceberg/TestDataTaskParser.java @@ -236,7 +236,7 @@ private DataTask createDataTask() { null)); return StaticDataTask.of( - Files.localInput("file:/tmp/metadata2.json"), + "/tmp/metadata2.json", SNAPSHOT_SCHEMA, SNAPSHOT_SCHEMA, snapshots, diff --git a/core/src/test/java/org/apache/iceberg/TestStaticDataTask.java b/core/src/test/java/org/apache/iceberg/TestStaticDataTask.java new file mode 100644 index 000000000000..b586887d291c --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestStaticDataTask.java @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg; + +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.List; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; + +public class TestStaticDataTask { + + private static final Schema TEST_SCHEMA = new Schema(required(1, "value", Types.LongType.get())); + + @Test + public void ofStringPathDoesNotReadFile() { + // Even a clearly-bogus path must be tolerated: the new overload promises no I/O against the + // location, only preserving it for identification. If the implementation regressed to building + // the DataFile via withInputFile (which calls InputFile.getLength()), this call would fail + // when the FileIO eventually tried to HEAD the path. + String bogusLocation = "file:///does/not/exist/never/HEADed.metadata.json"; + List values = ImmutableList.of(1L, 2L, 3L); + + DataTask task = + StaticDataTask.of( + bogusLocation, TEST_SCHEMA, TEST_SCHEMA, values, v -> StaticDataTask.Row.of(v)); + + assertThat(task.isDataTask()).as("should be a DataTask").isTrue(); + assertThat(task.length()).as("length must be 0 (no file bytes processed)").isEqualTo(0L); + assertThat(task.file().fileSizeInBytes()) + .as("DataFile size must be 0 (no HEAD on synthetic file)") + .isEqualTo(0L); + assertThat(task.file().format()) + .as("DataFile format must be METADATA") + .isEqualTo(FileFormat.METADATA); + assertThat(task.file().location()) + .as("DataFile path must preserve the supplied location for identification") + .isEqualTo(bogusLocation); + assertThat(task.file().recordCount()) + .as("DataFile record count must reflect the row count") + .isEqualTo(values.size()); + } + + @Test + public void ofStringPathRowsProjectCorrectly() { + String location = "file:///irrelevant/path.metadata.json"; + List values = ImmutableList.of(10L, 20L, 30L); + + DataTask task = + StaticDataTask.of( + location, TEST_SCHEMA, TEST_SCHEMA, values, v -> StaticDataTask.Row.of(v)); + + try (CloseableIterable rows = task.rows()) { + List read = + ImmutableList.copyOf( + org.apache.iceberg.relocated.com.google.common.collect.Iterables.transform( + rows, row -> row.get(0, Long.class))); + assertThat(read).containsExactlyElementsOf(values); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Test + public void ofStringPathWithNoValuesYieldsEmptyTask() { + String location = "file:///irrelevant/path.metadata.json"; + + DataTask task = + StaticDataTask.of( + location, + TEST_SCHEMA, + TEST_SCHEMA, + ImmutableList.of(), + v -> StaticDataTask.Row.of(v)); + + assertThat(task.length()).isEqualTo(0L); + assertThat(task.file().recordCount()).isEqualTo(0L); + + try (CloseableIterable rows = task.rows()) { + assertThat(rows).isEmpty(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } +}