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); + } + } +}