Repository navigation
Core: Fix locationProvider implementation for SerializableTable #12564
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
f1139d6
3c53461
9413990
99ee19b
01ee2e3
3c0507e
d0f51c0
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -30,6 +30,7 @@ | |
| import org.apache.iceberg.relocated.com.google.common.collect.Maps; | ||
| import org.apache.iceberg.util.SerializableMap; | ||
| import org.apache.iceberg.util.SerializableSupplier; | ||
| import org.apache.iceberg.util.TryUtil; | ||
|
|
||
| /** | ||
| * A read-only serializable table that can be sent to other nodes in a cluster. | ||
|
|
@@ -41,10 +42,10 @@ | |
| * table metadata, it directly persists the current schema, spec, sort order, table properties to | ||
| * avoid reading the metadata file from other nodes for frequently needed metadata. | ||
| * | ||
| * <p>The implementation assumes the passed instances of {@link FileIO}, {@link EncryptionManager} | ||
| * are serializable. If you are serializing the table using a custom serialization framework like | ||
| * Kryo, those instances of {@link FileIO}, {@link EncryptionManager} must be supported by that | ||
| * particular serialization framework. | ||
| * <p>The implementation assumes the passed instances of {@link FileIO}, {@link EncryptionManager}, | ||
| * {@link LocationProvider} are serializable. If you are serializing the table using a custom | ||
| * serialization framework like Kryo, those instances of {@link FileIO}, {@link EncryptionManager}, | ||
| * {@link LocationProvider} must be supported by that particular serialization framework. | ||
| * | ||
| * <p><em>Note:</em> loading the complete metadata from a large number of nodes can overwhelm the | ||
| * storage. | ||
|
|
@@ -65,8 +66,8 @@ public class SerializableTable implements Table, HasTableOperations, Serializabl | |
| private final Map<String, SnapshotRef> refs; | ||
| private final UUID uuid; | ||
| private final int formatVersion; | ||
| private final TryUtil.Try<LocationProvider> locationProviderTry; | ||
|
|
||
| private transient volatile LocationProvider lazyLocationProvider = null; | ||
| private transient volatile Table lazyTable = null; | ||
| private transient volatile Schema lazySchema = null; | ||
| private transient volatile Map<Integer, PartitionSpec> lazySpecs = null; | ||
|
|
@@ -85,6 +86,7 @@ protected SerializableTable(Table table) { | |
| this.sortOrderAsJson = SortOrderParser.toJson(table.sortOrder()); | ||
| this.io = fileIO(table); | ||
| this.encryption = table.encryption(); | ||
| this.locationProviderTry = TryUtil.run(table::locationProvider); | ||
| this.refs = SerializableMap.copyOf(table.refs()); | ||
| this.uuid = table.uuid(); | ||
| this.formatVersion = formatVersion(table); | ||
|
|
@@ -265,14 +267,15 @@ public EncryptionManager encryption() { | |
|
|
||
| @Override | ||
| public LocationProvider locationProvider() { | ||
| if (lazyLocationProvider == null) { | ||
| synchronized (this) { | ||
| if (lazyLocationProvider == null) { | ||
| this.lazyLocationProvider = LocationProviders.locationsFor(location, properties); | ||
| } | ||
| try { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We can move this logic into the Try class as |
||
| return this.locationProviderTry.get(); | ||
| } catch (Exception e) { | ||
| if (e instanceof RuntimeException) { | ||
| throw (RuntimeException) e; | ||
| } else { | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
| return lazyLocationProvider; | ||
| } | ||
|
|
||
| @Override | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,113 @@ | ||
| /* | ||
| * 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.util; | ||
|
|
||
| import java.io.Serializable; | ||
| import org.apache.iceberg.relocated.com.google.common.base.Preconditions; | ||
|
|
||
| /** Utility for executing code that may throw exceptions and deferring exception handling. */ | ||
| public class TryUtil { | ||
| private TryUtil() {} | ||
|
|
||
| /** | ||
| * Functional interface for operations that may throw exceptions. | ||
| * | ||
| * @param <T> return type of the operation | ||
| */ | ||
| @FunctionalInterface | ||
| public interface ThrowingSupplier<T> { | ||
| T get() throws Exception; | ||
| } | ||
|
|
||
| /** | ||
| * Executes the given operation and returns a Try object containing either the result or the | ||
| * exception. | ||
| * | ||
| * @param operation the operation to execute | ||
| * @param <T> the type of the result | ||
| * @return a Try object containing either the result or the exception | ||
| */ | ||
| public static <T> Try<T> run(ThrowingSupplier<T> operation) { | ||
| try { | ||
| return Try.success(operation.get()); | ||
| } catch (Exception e) { | ||
| return Try.failure(e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Container for the result of an operation that might throw an exception. | ||
| * | ||
| * @param <T> the type of the result | ||
| */ | ||
| public static class Try<T> implements Serializable { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Instead of using a util class what do you think about adding a static helper like: Then usage would look like
|
||
| private static final long serialVersionUID = 1L; | ||
|
|
||
| private final T value; | ||
| private final Exception exception; | ||
|
|
||
| private Try(T value, Exception exception) { | ||
| this.value = value; | ||
| this.exception = exception; | ||
| } | ||
|
|
||
| /** Creates a successful Try with the given value. */ | ||
| public static <T> Try<T> success(T value) { | ||
| return new Try<>(value, null); | ||
| } | ||
|
|
||
| /** Creates a failed Try with the given exception. */ | ||
| public static <T> Try<T> failure(Exception exception) { | ||
| Preconditions.checkNotNull(exception, "Exception cannot be null"); | ||
| return new Try<>(null, exception); | ||
| } | ||
|
|
||
| /** Checks if the operation was successful. */ | ||
| public boolean isSuccess() { | ||
| return exception == null; | ||
| } | ||
|
|
||
| /** Checks if the operation failed. */ | ||
| public boolean isFailure() { | ||
| return exception != null; | ||
| } | ||
|
|
||
| /** | ||
| * Gets the value if the operation was successful, or throws the original exception if it | ||
| * failed. | ||
| * | ||
| * @return the result value | ||
| * @throws Exception the original exception if the operation failed | ||
| */ | ||
| public T get() throws Exception { | ||
| if (exception != null) { | ||
| throw exception; | ||
| } | ||
| return value; | ||
| } | ||
|
|
||
| /** | ||
| * Gets the value if the operation was successful, or returns the provided default value if it | ||
| * failed. | ||
| */ | ||
| public T orElse(T defaultValue) { | ||
| return isSuccess() ? value : defaultValue; | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,158 @@ | ||||||
| /* | ||||||
| * 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.util; | ||||||
|
|
||||||
| import static org.assertj.core.api.Assertions.assertThat; | ||||||
| import static org.assertj.core.api.Assertions.assertThatThrownBy; | ||||||
|
|
||||||
| import java.io.ByteArrayInputStream; | ||||||
| import java.io.ByteArrayOutputStream; | ||||||
| import java.io.ObjectInputStream; | ||||||
| import java.io.ObjectOutputStream; | ||||||
| import org.junit.Test; | ||||||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||
|
|
||||||
| public class TestTryUtil { | ||||||
|
|
||||||
| @Test | ||||||
| public void testSuccessfulOperation() throws Exception { | ||||||
| TryUtil.Try<String> result = TryUtil.run(() -> "success"); | ||||||
|
|
||||||
| assertThat(result.isSuccess()).isTrue(); | ||||||
| assertThat(result.isFailure()).isFalse(); | ||||||
| assertThat(result.get()).isEqualTo("success"); | ||||||
| assertThat(result.orElse("default")).isEqualTo("success"); | ||||||
| } | ||||||
|
|
||||||
| @Test | ||||||
| public void testFailedOperation() { | ||||||
| Exception testException = new IllegalArgumentException("test exception"); | ||||||
| TryUtil.Try<String> result = | ||||||
| TryUtil.run( | ||||||
| () -> { | ||||||
| throw testException; | ||||||
| }); | ||||||
|
|
||||||
| assertThat(result.isSuccess()).isFalse(); | ||||||
| assertThat(result.isFailure()).isTrue(); | ||||||
| assertThat(result.orElse("default")).isEqualTo("default"); | ||||||
| } | ||||||
|
|
||||||
| @Test | ||||||
| public void testGetWithFailure() { | ||||||
| Exception testException = new IllegalArgumentException("test exception"); | ||||||
| TryUtil.Try<String> result = | ||||||
| TryUtil.run( | ||||||
| () -> { | ||||||
| throw testException; | ||||||
| }); | ||||||
|
|
||||||
| assertThatThrownBy(result::get).isSameAs(testException); | ||||||
| } | ||||||
|
|
||||||
| @Test | ||||||
| public void testCheckedException() { | ||||||
| TryUtil.Try<String> result = | ||||||
| TryUtil.run( | ||||||
| () -> { | ||||||
| throw new Exception("checked exception"); | ||||||
| }); | ||||||
|
|
||||||
| assertThat(result.isFailure()).isTrue(); | ||||||
| assertThatThrownBy(result::get).isInstanceOf(Exception.class).hasMessage("checked exception"); | ||||||
| } | ||||||
|
|
||||||
| @Test | ||||||
| public void testRuntimeException() { | ||||||
| RuntimeException runtimeException = new RuntimeException("runtime exception"); | ||||||
| TryUtil.Try<String> result = | ||||||
| TryUtil.run( | ||||||
| () -> { | ||||||
| throw runtimeException; | ||||||
| }); | ||||||
|
|
||||||
| assertThat(result.isFailure()).isTrue(); | ||||||
| assertThatThrownBy(result::get).isSameAs(runtimeException); | ||||||
| } | ||||||
|
|
||||||
| @Test | ||||||
| public void testNullValue() throws Exception { | ||||||
| TryUtil.Try<String> result = TryUtil.run(() -> null); | ||||||
|
|
||||||
| assertThat(result.isSuccess()).isTrue(); | ||||||
| assertThat(result.get()).isNull(); | ||||||
| assertThat(result.orElse("default")).isNull(); | ||||||
| } | ||||||
|
|
||||||
| @Test | ||||||
| public void testOrElseWithDefaultValue() { | ||||||
| TryUtil.Try<String> result = | ||||||
| TryUtil.run( | ||||||
| () -> { | ||||||
| throw new RuntimeException("test exception"); | ||||||
| }); | ||||||
|
|
||||||
| assertThat(result.orElse("default value")).isEqualTo("default value"); | ||||||
| } | ||||||
|
|
||||||
| @Test | ||||||
| public void testSerialization() throws Exception { | ||||||
| TryUtil.Try<String> original = TryUtil.run(() -> "serialized value"); | ||||||
|
|
||||||
| // Serialize to byte array | ||||||
| ByteArrayOutputStream bytes = new ByteArrayOutputStream(); | ||||||
| ObjectOutputStream out = new ObjectOutputStream(bytes); | ||||||
| out.writeObject(original); | ||||||
| out.close(); | ||||||
|
|
||||||
| // Deserialize from byte array | ||||||
| ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray())); | ||||||
| TryUtil.Try<String> deserialized = (TryUtil.Try<String>) in.readObject(); | ||||||
| in.close(); | ||||||
|
|
||||||
| // Verify state is preserved | ||||||
| assertThat(deserialized.isSuccess()).isTrue(); | ||||||
| assertThat(deserialized.get()).isEqualTo("serialized value"); | ||||||
| } | ||||||
|
|
||||||
| @Test | ||||||
| public void testSerializationWithException() throws Exception { | ||||||
| Exception originalException = new IllegalArgumentException("test exception"); | ||||||
| TryUtil.Try<String> original = | ||||||
| TryUtil.run( | ||||||
| () -> { | ||||||
| throw originalException; | ||||||
| }); | ||||||
| // Serialize to byte array | ||||||
| ByteArrayOutputStream bytes = new ByteArrayOutputStream(); | ||||||
| ObjectOutputStream out = new ObjectOutputStream(bytes); | ||||||
| out.writeObject(original); | ||||||
| out.close(); | ||||||
|
|
||||||
| // Deserialize from byte array | ||||||
| ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray())); | ||||||
| TryUtil.Try<String> deserialized = (TryUtil.Try<String>) in.readObject(); | ||||||
| in.close(); | ||||||
|
|
||||||
| // Verify state is preserved | ||||||
| assertThat(deserialized.isFailure()).isTrue(); | ||||||
| assertThatThrownBy(deserialized::get) | ||||||
| .isInstanceOf(IllegalArgumentException.class) | ||||||
| .hasMessage("test exception"); | ||||||
| } | ||||||
| } | ||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we add some tests in spark to ensure that the exception isn't thrown on instantiation but when
serializableTable.locationProvider()is thrown