advancedxy commented on code in PR #10755: URL: https://github.com/apache/iceberg/pull/10755#discussion_r1696133057
########## core/src/main/java/org/apache/iceberg/BaseRemoveUnusedSpecs.java: ########## @@ -0,0 +1,103 @@ +/* + * 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.TableProperties.COMMIT_MAX_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT; + +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.Tasks; + +/** + * Implementation of RemoveUnusedSpecs API to remove unused partition specs. + * + * <p>When committing, these changes will be applied to the latest table metadata. Commit conflicts + * will be resolved by recalculating which specs are no longer in use again in the latest metadata + * and retrying. + */ +class BaseRemoveUnusedSpecs implements RemoveUnusedSpecs { + private final TableOperations ops; + private final Table table; + + BaseRemoveUnusedSpecs(TableOperations ops, Table table) { + this.ops = ops; + this.table = table; + } + + @Override + public List<PartitionSpec> apply() { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + return newMetadata.specs(); + } + + @Override + public void commit() { + TableMetadata base = ops.refresh(); + Tasks.foreach(ops) + .retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT)) + .exponentialBackoff( + base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT), + 2.0 /* exponential */) + .onlyRetryOn(CommitFailedException.class) + .run( + taskOps -> { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + taskOps.commit(current, newMetadata); + }); + } + + private TableMetadata removeUnusedSpecs(TableMetadata current) { + List<PartitionSpec> specs = current.specs(); + int currentSpecId = current.defaultSpecId(); + + // Read ManifestLists and get all specId's in use + Set<Integer> specsInUse = + Sets.newHashSet( + CloseableIterable.transform( + MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.ALL_ENTRIES) + .newScan() + .planFiles(), + task -> ((BaseEntriesTable.ManifestReadTask) task).partitionSpecId())); + + // add current spec id to the set of specs in use Review Comment: Will remove it. ########## core/src/main/java/org/apache/iceberg/TableMetadata.java: ########## @@ -597,6 +597,24 @@ public TableMetadata replaceProperties(Map<String, String> rawProperties) { .build(); } + /** + * Prune the unused partition specs from the table metadata. + * + * <p>Note: it's not safe for external client to call this directly, it's usually called by the Review Comment: Let me remove the note then. > Also, we are no longer adding new operation methods to this. These days we make modifications to the Builder instead. For this part, I think modifications are still going through the Builders. The main purpose of this method is to provide a single access point to prune unused specs purely so that prune unused specs are not mixed with other metadata updates. ########## core/src/main/java/org/apache/iceberg/TableMetadata.java: ########## @@ -1425,6 +1460,7 @@ private boolean hasChanges() { || (discardChanges && !changes.isEmpty()) || metadataLocation != null || suppressHistoricalSnapshots + || hasRemovedSpecs Review Comment: > is this a way so that we avoid having to add the metadata update type for REST (since then that would ultimately just be in the changes list)? Yes, as discussed earlier, I don't think we should expose `removePartitionSpec` directly to the REST API as there's no easy way to ensure that the spec to remove is indeed not used. > This is not the right way to update hasChanges. Instead, this needs to add a change to changes. It was adding a `RemovePartitionSpec` to the changes. However, that would require a REST spec change and it's a bit heavy. See discussions as well: https://github.com/apache/iceberg/pull/10755#issuecomment-2246811037 > That way it is sent to REST services to modify the table in the REST commit path. I might be missing something, so all the metadata changes have to be sent to the REST service? I didn't work with a REST catalog before and don't see how changes are sent to the REST service. It would be great that some reference or code could be pointed to. ########## api/src/main/java/org/apache/iceberg/Table.java: ########## @@ -211,6 +211,17 @@ default IncrementalChangelogScan newIncrementalChangelogScan() { */ AppendFiles newAppend(); + /** + * Remove any partition specs from the Metadata that are no longer used in any data files. Always + * preserves the current default spec even if it has not yet been used. Review Comment: Sounds better to me, will fix it in a new commit. ########## core/src/main/java/org/apache/iceberg/TableMetadata.java: ########## @@ -1446,6 +1482,11 @@ public TableMetadata build() { "Cannot set metadata location with changes to table metadata: %s changes", changes.size()); + if (hasRemovedSpecs) { + Preconditions.checkArgument( + changes.isEmpty(), "Cannot remove partition specs with other metadata update"); Review Comment: See https://github.com/apache/iceberg/pull/10755/files#r1690420193 and https://github.com/apache/iceberg/pull/10755/files#r1696099907 I think this check is to make sure that `RemoveUnusedSpecs` and other metadata updates are not happened together. ########## core/src/main/java/org/apache/iceberg/TableMetadata.java: ########## @@ -1102,6 +1121,22 @@ public Builder setDefaultPartitionSpec(int specId) { return this; } + private Builder removePartitionSpec(PartitionSpec spec) { + Preconditions.checkArgument( + changes.isEmpty(), "Cannot remove partition spec with other metadata update"); Review Comment: > I think this is needed to avoid conflicts with changes to the default spec ID. Yeah, besides `SetDefaultSepc`, `AddPartitionSpec` might also interfere with this. For safety purposes, the previous implementation rejects all other metadata updates since `removePartitionSpec` is rarely called and always called alone. ########## core/src/main/java/org/apache/iceberg/TableMetadata.java: ########## @@ -1102,6 +1121,22 @@ public Builder setDefaultPartitionSpec(int specId) { return this; } + private Builder removePartitionSpec(PartitionSpec spec) { + Preconditions.checkArgument( + changes.isEmpty(), "Cannot remove partition spec with other metadata update"); + Preconditions.checkArgument( + spec.specId() != defaultSpecId, "Cannot remove default partition spec"); + PartitionSpec toBeRemoved = specsById.remove(spec.specId()); + Preconditions.checkArgument( + toBeRemoved != null && toBeRemoved.equals(spec), + "Cannot remove an unknown spec, spec id: %s", + spec.specId()); + this.specs = Review Comment: > There's no need to fail something if it has an accidental retry after the first attempt succeeded. This is a very good point, let me reconsider this part. ########## core/src/main/java/org/apache/iceberg/BaseRemoveUnusedSpecs.java: ########## @@ -0,0 +1,103 @@ +/* + * 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.TableProperties.COMMIT_MAX_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT; + +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.Tasks; + +/** + * Implementation of RemoveUnusedSpecs API to remove unused partition specs. + * + * <p>When committing, these changes will be applied to the latest table metadata. Commit conflicts + * will be resolved by recalculating which specs are no longer in use again in the latest metadata + * and retrying. + */ +class BaseRemoveUnusedSpecs implements RemoveUnusedSpecs { + private final TableOperations ops; + private final Table table; + + BaseRemoveUnusedSpecs(TableOperations ops, Table table) { + this.ops = ops; + this.table = table; + } + + @Override + public List<PartitionSpec> apply() { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + return newMetadata.specs(); + } + + @Override + public void commit() { + TableMetadata base = ops.refresh(); + Tasks.foreach(ops) + .retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT)) + .exponentialBackoff( + base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT), + 2.0 /* exponential */) + .onlyRetryOn(CommitFailedException.class) + .run( + taskOps -> { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + taskOps.commit(current, newMetadata); + }); + } + + private TableMetadata removeUnusedSpecs(TableMetadata current) { + List<PartitionSpec> specs = current.specs(); + int currentSpecId = current.defaultSpecId(); + + // Read ManifestLists and get all specId's in use + Set<Integer> specsInUse = + Sets.newHashSet( + CloseableIterable.transform( + MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.ALL_ENTRIES) + .newScan() + .planFiles(), + task -> ((BaseEntriesTable.ManifestReadTask) task).partitionSpecId())); Review Comment: Good point, let me refactor this to avoid unchecked cast. > If I understand correctly, the purpose of using a metadata table here is not to use the Table interface, but instead to reuse some of the code in the all_entries table I think this code is used to avoid actually accessing the underlying rows, see https://github.com/apache/iceberg/pull/3462#discussion_r742310413. We can/should use the `Table` interface to access Manifest files directly. How does that sound to you? ########## core/src/main/java/org/apache/iceberg/BaseRemoveUnusedSpecs.java: ########## @@ -0,0 +1,103 @@ +/* + * 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.TableProperties.COMMIT_MAX_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT; + +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.Tasks; + +/** + * Implementation of RemoveUnusedSpecs API to remove unused partition specs. + * + * <p>When committing, these changes will be applied to the latest table metadata. Commit conflicts + * will be resolved by recalculating which specs are no longer in use again in the latest metadata + * and retrying. + */ +class BaseRemoveUnusedSpecs implements RemoveUnusedSpecs { + private final TableOperations ops; + private final Table table; + + BaseRemoveUnusedSpecs(TableOperations ops, Table table) { + this.ops = ops; + this.table = table; + } + + @Override + public List<PartitionSpec> apply() { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + return newMetadata.specs(); + } + + @Override + public void commit() { + TableMetadata base = ops.refresh(); + Tasks.foreach(ops) + .retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT)) + .exponentialBackoff( + base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT), + 2.0 /* exponential */) + .onlyRetryOn(CommitFailedException.class) + .run( + taskOps -> { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + taskOps.commit(current, newMetadata); + }); + } + + private TableMetadata removeUnusedSpecs(TableMetadata current) { + List<PartitionSpec> specs = current.specs(); + int currentSpecId = current.defaultSpecId(); + + // Read ManifestLists and get all specId's in use + Set<Integer> specsInUse = + Sets.newHashSet( + CloseableIterable.transform( + MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.ALL_ENTRIES) + .newScan() + .planFiles(), + task -> ((BaseEntriesTable.ManifestReadTask) task).partitionSpecId())); + + // add current spec id to the set of specs in use + specsInUse.add(currentSpecId); + + List<PartitionSpec> toRemoveSpecs = Review Comment: Nice suggestion, will rename it. ########## core/src/main/java/org/apache/iceberg/BaseRemoveUnusedSpecs.java: ########## @@ -0,0 +1,103 @@ +/* + * 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.TableProperties.COMMIT_MAX_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT; + +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.Tasks; + +/** + * Implementation of RemoveUnusedSpecs API to remove unused partition specs. + * + * <p>When committing, these changes will be applied to the latest table metadata. Commit conflicts + * will be resolved by recalculating which specs are no longer in use again in the latest metadata + * and retrying. + */ +class BaseRemoveUnusedSpecs implements RemoveUnusedSpecs { + private final TableOperations ops; + private final Table table; + + BaseRemoveUnusedSpecs(TableOperations ops, Table table) { + this.ops = ops; + this.table = table; + } + + @Override + public List<PartitionSpec> apply() { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + return newMetadata.specs(); + } + + @Override + public void commit() { + TableMetadata base = ops.refresh(); + Tasks.foreach(ops) + .retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT)) + .exponentialBackoff( + base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT), + 2.0 /* exponential */) + .onlyRetryOn(CommitFailedException.class) + .run( + taskOps -> { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + taskOps.commit(current, newMetadata); + }); + } + + private TableMetadata removeUnusedSpecs(TableMetadata current) { Review Comment: Let me rename it to `internalApply` then. ########## core/src/main/java/org/apache/iceberg/BaseRemoveUnusedSpecs.java: ########## @@ -0,0 +1,103 @@ +/* + * 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.TableProperties.COMMIT_MAX_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT; + +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.Tasks; + +/** + * Implementation of RemoveUnusedSpecs API to remove unused partition specs. + * + * <p>When committing, these changes will be applied to the latest table metadata. Commit conflicts + * will be resolved by recalculating which specs are no longer in use again in the latest metadata + * and retrying. + */ +class BaseRemoveUnusedSpecs implements RemoveUnusedSpecs { + private final TableOperations ops; + private final Table table; + + BaseRemoveUnusedSpecs(TableOperations ops, Table table) { + this.ops = ops; + this.table = table; + } + + @Override + public List<PartitionSpec> apply() { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + return newMetadata.specs(); + } + + @Override + public void commit() { + TableMetadata base = ops.refresh(); + Tasks.foreach(ops) + .retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT)) + .exponentialBackoff( + base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT), + 2.0 /* exponential */) + .onlyRetryOn(CommitFailedException.class) + .run( + taskOps -> { + TableMetadata current = ops.refresh(); + TableMetadata newMetadata = removeUnusedSpecs(current); + taskOps.commit(current, newMetadata); + }); + } + + private TableMetadata removeUnusedSpecs(TableMetadata current) { + List<PartitionSpec> specs = current.specs(); + int currentSpecId = current.defaultSpecId(); + + // Read ManifestLists and get all specId's in use Review Comment: Thanks, let me remove it. -- 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-unsubscr...@iceberg.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org For additional commands, e-mail: issues-h...@iceberg.apache.org
