diff --git a/core/src/main/java/org/apache/iceberg/MetadataUpdate.java b/core/src/main/java/org/apache/iceberg/MetadataUpdate.java index d133b76901da..7c7878aecca4 100644 --- a/core/src/main/java/org/apache/iceberg/MetadataUpdate.java +++ b/core/src/main/java/org/apache/iceberg/MetadataUpdate.java @@ -19,6 +19,7 @@ package org.apache.iceberg; import java.io.Serializable; +import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Set; @@ -490,4 +491,21 @@ public void applyTo(ViewMetadata.Builder viewMetadataBuilder) { viewMetadataBuilder.setCurrentVersionId(versionId); } } + + class AppendFilesUpdate implements MetadataUpdate { + private final List addedManifests; + + public AppendFilesUpdate(List addedManifests) { + this.addedManifests = addedManifests; + } + + public List getAddedManifests() { + return addedManifests; + } + + @Override + public void applyTo(TableMetadata.Builder tableMetadataBuilder) { + tableMetadataBuilder.appendFiles(addedManifests); + } + } } diff --git a/core/src/main/java/org/apache/iceberg/MetadataUpdateParser.java b/core/src/main/java/org/apache/iceberg/MetadataUpdateParser.java index 8cdfd3c72b6e..194a4df731f4 100644 --- a/core/src/main/java/org/apache/iceberg/MetadataUpdateParser.java +++ b/core/src/main/java/org/apache/iceberg/MetadataUpdateParser.java @@ -21,6 +21,7 @@ import com.fasterxml.jackson.core.JsonGenerator; import com.fasterxml.jackson.databind.JsonNode; import java.io.IOException; +import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Set; @@ -59,6 +60,7 @@ private MetadataUpdateParser() {} static final String SET_CURRENT_VIEW_VERSION = "set-current-view-version"; static final String SET_PARTITION_STATISTICS = "set-partition-statistics"; static final String REMOVE_PARTITION_STATISTICS = "remove-partition-statistics"; + static final String APPEND_FILES = "append-files"; // AssignUUID private static final String UUID = "uuid"; @@ -126,6 +128,9 @@ private MetadataUpdateParser() {} // SetCurrentViewVersion private static final String VIEW_VERSION_ID = "view-version-id"; + // Data operations + private static final String APPENDED_MANIFESTS = "appended-manifests"; + private static final Map, String> ACTIONS = ImmutableMap., String>builder() .put(MetadataUpdate.AssignUUID.class, ASSIGN_UUID) @@ -149,6 +154,7 @@ private MetadataUpdateParser() {} .put(MetadataUpdate.SetLocation.class, SET_LOCATION) .put(MetadataUpdate.AddViewVersion.class, ADD_VIEW_VERSION) .put(MetadataUpdate.SetCurrentViewVersion.class, SET_CURRENT_VIEW_VERSION) + .put(MetadataUpdate.AppendFilesUpdate.class, APPEND_FILES) .buildOrThrow(); public static String toJson(MetadataUpdate metadataUpdate) { @@ -241,6 +247,9 @@ public static void toJson(MetadataUpdate metadataUpdate, JsonGenerator generator writeSetCurrentViewVersionId( (MetadataUpdate.SetCurrentViewVersion) metadataUpdate, generator); break; + case APPEND_FILES: + writeAppendFiles((MetadataUpdate.AppendFilesUpdate) metadataUpdate, generator); + break; default: throw new IllegalArgumentException( String.format( @@ -312,6 +321,8 @@ public static MetadataUpdate fromJson(JsonNode jsonNode) { return readAddViewVersion(jsonNode); case SET_CURRENT_VIEW_VERSION: return readCurrentViewVersionId(jsonNode); + case APPEND_FILES: + return readAppendFiles(jsonNode); default: throw new UnsupportedOperationException( String.format("Cannot convert metadata update action to json: %s", action)); @@ -346,6 +357,13 @@ private static void writeAddPartitionSpec( PartitionSpecParser.toJson(update.spec(), gen); } + private static void writeAppendFiles(MetadataUpdate.AppendFilesUpdate update, JsonGenerator gen) + throws IOException { + if (update.getAddedManifests() != null) { + JsonUtil.writeStringArray(APPENDED_MANIFESTS, update.getAddedManifests(), gen); + } + } + private static void writeSetDefaultPartitionSpec( MetadataUpdate.SetDefaultPartitionSpec update, JsonGenerator gen) throws IOException { gen.writeNumberField(SPEC_ID, update.specId()); @@ -452,6 +470,11 @@ private static MetadataUpdate readAssignUUID(JsonNode node) { return new MetadataUpdate.AssignUUID(uuid); } + private static MetadataUpdate readAppendFiles(JsonNode node) { + List metadataLocations = JsonUtil.getStringList(APPENDED_MANIFESTS, node); + return new MetadataUpdate.AppendFilesUpdate(metadataLocations); + } + private static MetadataUpdate readUpgradeFormatVersion(JsonNode node) { int formatVersion = JsonUtil.getInt(FORMAT_VERSION, node); return new MetadataUpdate.UpgradeFormatVersion(formatVersion); diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java index b9061a3107ac..b18e66af20a9 100644 --- a/core/src/main/java/org/apache/iceberg/TableMetadata.java +++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java @@ -1779,5 +1779,10 @@ private boolean isAddedSnapshot(long snapshotId) { private Stream changes(Class updateClass) { return changes.stream().filter(updateClass::isInstance).map(updateClass::cast); } + + public Builder appendFiles(List files) { + changes.add(new MetadataUpdate.AppendFilesUpdate(files)); + return this; + } } } diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java index 5f660f0f4fe8..ed815501ccf4 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java @@ -35,7 +35,6 @@ import java.util.function.BiFunction; import java.util.function.Function; import java.util.function.Supplier; -import org.apache.iceberg.BaseTable; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.EnvironmentContext; @@ -112,6 +111,7 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog private static final Logger LOG = LoggerFactory.getLogger(RESTSessionCatalog.class); private static final String DEFAULT_FILE_IO_IMPL = "org.apache.iceberg.io.ResolvingFileIO"; private static final String REST_METRICS_REPORTING_ENABLED = "rest-metrics-reporting-enabled"; + private static final String REST_DATA_COMMIT_ENABLED = "rest-data-commit-enabled"; private static final String REST_SNAPSHOT_LOADING_MODE = "snapshot-loading-mode"; private static final List TOKEN_PREFERENCE_ORDER = ImmutableList.of( @@ -135,6 +135,7 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog private FileIO io = null; private MetricsReporter reporter = null; private boolean reportingViaRestEnabled; + private boolean dataCommitViaRestEnabled; private CloseableGroup closeables = null; // a lazy thread pool for token refresh @@ -219,7 +220,8 @@ public void initialize(String name, Map unresolved) { AuthSession.fromAccessToken( client, tokenRefreshExecutor(), token, expiresAtMillis(mergedProps), catalogAuth); } - + this.dataCommitViaRestEnabled = + PropertyUtil.propertyAsBoolean(mergedProps, REST_DATA_COMMIT_ENABLED, false); this.io = newFileIO(SessionContext.createEmpty(), mergedProps); this.fileIOCloser = newFileIOCloser(); @@ -391,12 +393,15 @@ public Table loadTable(SessionContext context, TableIdentifier identifier) { tableMetadata); trackFileIO(ops); - - BaseTable table = - new BaseTable( + Table table = + new RESTTable( ops, fullTableName(finalIdentifier), - metricsReporter(paths.metrics(finalIdentifier), session::headers)); + metricsReporter(paths.metrics(finalIdentifier), session::headers), + this.client, + paths.table(finalIdentifier), + session::headers, + dataCommitViaRestEnabled); if (metadataType != null) { return MetadataTableUtils.createMetadataTableInstance(table, metadataType); } @@ -464,9 +469,14 @@ public Table registerTable( response.tableMetadata()); trackFileIO(ops); - - return new BaseTable( - ops, fullTableName(ident), metricsReporter(paths.metrics(ident), session::headers)); + return new RESTTable( + ops, + fullTableName(ident), + metricsReporter(paths.metrics(ident), session::headers), + client, + paths.table(ident), + session::headers, + dataCommitViaRestEnabled); } @Override @@ -683,9 +693,14 @@ public Table create() { response.tableMetadata()); trackFileIO(ops); - - return new BaseTable( - ops, fullTableName(ident), metricsReporter(paths.metrics(ident), session::headers)); + return new RESTTable( + ops, + fullTableName(ident), + metricsReporter(paths.metrics(ident), session::headers), + client, + paths.table(ident), + session::headers, + dataCommitViaRestEnabled); } @Override diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTTable.java b/core/src/main/java/org/apache/iceberg/rest/RESTTable.java new file mode 100644 index 000000000000..ae30cd447198 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/rest/RESTTable.java @@ -0,0 +1,59 @@ +/* + * + * * 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.rest; + +import java.util.Map; +import java.util.function.Supplier; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.metrics.MetricsReporter; +import org.apache.iceberg.rest.operations.RestAppendFiles; + +public class RESTTable extends BaseTable { + private final RESTClient client; + private final String path; + private final Supplier> headers; + private final boolean dataCommitViaRestEnabled; + + public RESTTable( + TableOperations ops, + String name, + MetricsReporter reporter, + RESTClient client, + String path, + Supplier> headers, + boolean dataCommitViaRestEnabled) { + super(ops, name, reporter); + this.client = client; + this.headers = headers; + this.path = path; + this.dataCommitViaRestEnabled = dataCommitViaRestEnabled; + } + + @Override + public AppendFiles newAppend() { + if (dataCommitViaRestEnabled) { + return new RestAppendFiles(client, path, headers, operations()); + } + return super.newAppend(); + } +} diff --git a/core/src/main/java/org/apache/iceberg/rest/operations/RestAppendFiles.java b/core/src/main/java/org/apache/iceberg/rest/operations/RestAppendFiles.java new file mode 100644 index 000000000000..2488f348e369 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/rest/operations/RestAppendFiles.java @@ -0,0 +1,190 @@ +/* + * + * * 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.rest.operations; + +import static org.apache.iceberg.TableProperties.MANIFEST_TARGET_SIZE_BYTES; +import static org.apache.iceberg.TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; +import java.util.function.Supplier; +import java.util.stream.Collectors; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestReader; +import org.apache.iceberg.ManifestWriter; +import org.apache.iceberg.MetadataUpdate; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.RollingManifestWriter; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.UpdateRequirement; +import org.apache.iceberg.UpdateRequirements; +import org.apache.iceberg.exceptions.RuntimeIOException; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.rest.ErrorHandlers; +import org.apache.iceberg.rest.RESTClient; +import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.LoadTableResponse; + +public class RestAppendFiles implements AppendFiles { + private final RESTClient client; + private final String path; + private final Supplier> headers; + private final TableOperations ops; + private final PartitionSpec spec; + private final long targetManifestSizeBytes; + private final String commitUUID = UUID.randomUUID().toString(); + private final AtomicInteger manifestCount = new AtomicInteger(0); + private volatile Long snapshotId = null; + private final List newDataFiles = Lists.newArrayList(); + + public RestAppendFiles( + RESTClient client, String path, Supplier> headers, TableOperations ops) { + this.client = client; + this.path = path; + this.headers = headers; + this.ops = ops; + this.spec = ops.current().spec(); + this.targetManifestSizeBytes = + ops.current() + .propertyAsLong(MANIFEST_TARGET_SIZE_BYTES, MANIFEST_TARGET_SIZE_BYTES_DEFAULT); + } + + @Override + public void commit() { + MetadataUpdate.AppendFilesUpdate appendFilesUpdate = constructMetadataUpdate(); + List requirements = + UpdateRequirements.forUpdateTable(ops.current(), ImmutableList.of(appendFilesUpdate)); + UpdateTableRequest request = + new UpdateTableRequest(requirements, ImmutableList.of(appendFilesUpdate)); + client.post( + path, request, LoadTableResponse.class, headers, ErrorHandlers.tableCommitHandler()); + } + + private MetadataUpdate.AppendFilesUpdate constructMetadataUpdate() { + List addedManifests = constructManifests(); + return new MetadataUpdate.AppendFilesUpdate(addedManifests); + } + + @Override + public AppendFiles appendFile(DataFile file) { + newDataFiles.add(file); + return this; + } + + @Override + public RestAppendFiles appendManifest(ManifestFile manifest) { + Preconditions.checkArgument( + !manifest.hasExistingFiles(), "Cannot append manifest with existing files"); + Preconditions.checkArgument( + !manifest.hasDeletedFiles(), "Cannot append manifest with deleted files"); + Preconditions.checkArgument( + manifest.snapshotId() == null || manifest.snapshotId() == -1, + "Snapshot id must be assigned during commit"); + Preconditions.checkArgument( + manifest.sequenceNumber() == -1, "Sequence number must be assigned during commit"); + + // append data files from the manifest + ManifestReader reader = ManifestFiles.read(manifest, ops.io()); + reader.forEach(this::appendFile); + return this; + } + + private List constructManifests() { + List manifests = Lists.newArrayList(); + try { + RollingManifestWriter writer = newRollingManifestWriter(); + try { + newDataFiles.forEach(writer::add); + } finally { + writer.close(); + } + manifests.addAll(writer.toManifestFiles()); + } catch (IOException e) { + throw new RuntimeIOException(e, "Failed to write manifest"); + } + return manifests.stream().map(ManifestFile::path).collect(Collectors.toList()); + } + + protected RollingManifestWriter newRollingManifestWriter() { + return new RollingManifestWriter<>(this::newManifestWriter, targetManifestSizeBytes); + } + + protected ManifestWriter newManifestWriter() { + return ManifestFiles.write( + ops.current().formatVersion(), spec, newManifestOutput(), snapshotId()); + } + + protected OutputFile newManifestOutput() { + return ops.io() + .newOutputFile( + ops.metadataFileLocation( + FileFormat.AVRO.addExtension(commitUUID + "-m" + manifestCount.getAndIncrement()))); + } + + protected long snapshotId() { + if (snapshotId == null) { + synchronized (this) { + while (snapshotId == null || ops.current().snapshot(snapshotId) != null) { + this.snapshotId = ops.newSnapshotId(); + } + } + } + return snapshotId; + } + + @Override + public AppendFiles set(String property, String value) { + return this; + } + + @Override + public AppendFiles deleteWith(Consumer deleteFunc) { + return null; + } + + @Override + public AppendFiles stageOnly() { + return null; + } + + @Override + public AppendFiles scanManifestsWith(ExecutorService executorService) { + return null; + } + + @Override + public Snapshot apply() { + return null; + } +} diff --git a/open-api/rest-catalog-open-api.py b/open-api/rest-catalog-open-api.py index cc70d6d4cd89..02b15c6a5d87 100644 --- a/open-api/rest-catalog-open-api.py +++ b/open-api/rest-catalog-open-api.py @@ -328,6 +328,15 @@ class RemovePropertiesUpdate(BaseUpdate): removals: List[str] +class AppendFilesUpdate(BaseUpdate): + action: Optional[Literal['append-files']] = None + appended_manifests: List[str] = Field( + ..., + alias='appended-manifests', + description='Manifest files of DataFiles appended to a table', + ) + + class AddViewVersionUpdate(BaseUpdate): action: Literal['add-view-version'] view_version: ViewVersion = Field(..., alias='view-version') @@ -758,6 +767,7 @@ class TableUpdate(BaseModel): SetLocationUpdate, SetPropertiesUpdate, RemovePropertiesUpdate, + AppendFilesUpdate, ] diff --git a/open-api/rest-catalog-open-api.yaml b/open-api/rest-catalog-open-api.yaml index f0819a189817..bf9db113f100 100644 --- a/open-api/rest-catalog-open-api.yaml +++ b/open-api/rest-catalog-open-api.yaml @@ -2150,6 +2150,7 @@ components: remove-properties: '#/components/schemas/RemovePropertiesUpdate' add-view-version: '#/components/schemas/AddViewVersionUpdate' set-current-view-version: '#/components/schemas/SetCurrentViewVersionUpdate' + append-files: '#/components/schemas/AppendFilesUpdate' type: object required: - action @@ -2367,6 +2368,21 @@ components: items: type: string + AppendFilesUpdate: + allOf: + - $ref: '#/components/schemas/BaseUpdate' + required: + - appended-manifests + properties: + action: + type: string + enum: [ "append-files" ] + appended-manifests: + type: array + items: + type: string + description: Manifest files of DataFiles appended to a table + AddViewVersionUpdate: allOf: - $ref: '#/components/schemas/BaseUpdate' @@ -2411,6 +2427,7 @@ components: - $ref: '#/components/schemas/SetLocationUpdate' - $ref: '#/components/schemas/SetPropertiesUpdate' - $ref: '#/components/schemas/RemovePropertiesUpdate' + - $ref: '#/components/schemas/AppendFilesUpdate' ViewUpdate: anyOf: