This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 75862a8a9a Core: Add tests for lazy snapshot history loading (#8489)
75862a8a9a is described below
commit 75862a8a9af17e30890a4f19f93d9684cf212986
Author: Eduard Tudenhoefner <[email protected]>
AuthorDate: Tue Sep 12 18:05:58 2023 +0200
Core: Add tests for lazy snapshot history loading (#8489)
---
.../org/apache/iceberg/rest/TestRESTCatalog.java | 94 +++++++++++++++++-----
1 file changed, 73 insertions(+), 21 deletions(-)
diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
index a45f23b4b4..fe708bde11 100644
--- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
@@ -34,12 +34,10 @@ import java.io.IOException;
import java.nio.file.Path;
import java.util.Map;
import java.util.Optional;
-import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
-import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.BaseTransaction;
import org.apache.iceberg.CatalogProperties;
@@ -49,7 +47,6 @@ import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
-import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.Transaction;
@@ -814,16 +811,10 @@ public class TestRESTCatalog extends
CatalogTests<RESTCatalog> {
Answer<?> refsAnswer =
invocation -> {
LoadTableResponse originalResponse = (LoadTableResponse)
invocation.callRealMethod();
- TableMetadata fullTableMetadata = originalResponse.tableMetadata();
-
- Set<Long> referencedSnapshotIds =
- fullTableMetadata.refs().values().stream()
- .map(SnapshotRef::snapshotId)
- .collect(Collectors.toSet());
-
TableMetadata refsMetadata =
- fullTableMetadata.removeSnapshotsIf(
- s -> !referencedSnapshotIds.contains(s.snapshotId()));
+ TableMetadata.buildFrom(originalResponse.tableMetadata())
+ .suppressHistoricalSnapshots()
+ .build();
return LoadTableResponse.builder()
.withTableMetadata(refsMetadata)
@@ -924,16 +915,10 @@ public class TestRESTCatalog extends
CatalogTests<RESTCatalog> {
Answer<?> refsAnswer =
invocation -> {
LoadTableResponse originalResponse = (LoadTableResponse)
invocation.callRealMethod();
- TableMetadata fullTableMetadata = originalResponse.tableMetadata();
-
- Set<Long> referencedSnapshotIds =
- fullTableMetadata.refs().values().stream()
- .map(SnapshotRef::snapshotId)
- .collect(Collectors.toSet());
-
TableMetadata refsMetadata =
- fullTableMetadata.removeSnapshotsIf(
- s -> !referencedSnapshotIds.contains(s.snapshotId()));
+ TableMetadata.buildFrom(originalResponse.tableMetadata())
+ .suppressHistoricalSnapshots()
+ .build();
return LoadTableResponse.builder()
.withTableMetadata(refsMetadata)
@@ -996,6 +981,73 @@ public class TestRESTCatalog extends
CatalogTests<RESTCatalog> {
.hasSizeGreaterThan(Lists.newArrayList(table.snapshots()).size());
}
+ @Test
+ public void lazySnapshotLoadingWithDivergedHistory() {
+ RESTCatalogAdapter adapter = Mockito.spy(new
RESTCatalogAdapter(backendCatalog));
+
+ RESTCatalog catalog =
+ new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config)
-> adapter);
+ catalog.initialize(
+ "test",
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ "ignored",
+ CatalogProperties.FILE_IO_IMPL,
+ "org.apache.iceberg.inmemory.InMemoryFileIO",
+ "snapshot-loading-mode",
+ "refs"));
+
+ Table table =
+ catalog.createTable(TABLE, SCHEMA, PartitionSpec.unpartitioned(),
ImmutableMap.of());
+
+ int numSnapshots = 5;
+
+ for (int i = 0; i < numSnapshots; i++) {
+ table
+ .newFastAppend()
+ .appendFile(
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(String.format("/path/to/data-%s.parquet", i))
+ .withFileSizeInBytes(10)
+ .withRecordCount(2)
+ .build())
+ .commit();
+ }
+
+ ResourcePaths paths =
ResourcePaths.forCatalogProperties(Maps.newHashMap());
+
+ // Respond with only referenced snapshots
+ Answer<?> refsAnswer =
+ invocation -> {
+ LoadTableResponse originalResponse = (LoadTableResponse)
invocation.callRealMethod();
+ TableMetadata refsMetadata =
+ TableMetadata.buildFrom(originalResponse.tableMetadata())
+ .suppressHistoricalSnapshots()
+ .build();
+
+ return LoadTableResponse.builder()
+ .withTableMetadata(refsMetadata)
+ .addAllConfig(originalResponse.config())
+ .build();
+ };
+
+ Mockito.doAnswer(refsAnswer)
+ .when(adapter)
+ .execute(
+ eq(HTTPMethod.GET),
+ eq(paths.table(TABLE)),
+ eq(ImmutableMap.of("snapshots", "refs")),
+ any(),
+ eq(LoadTableResponse.class),
+ any(),
+ any());
+
+ Table refsTables = catalog.loadTable(TABLE);
+
assertThat(refsTables.currentSnapshot()).isEqualTo(table.currentSnapshot());
+ assertThat(refsTables.snapshots()).hasSize(numSnapshots);
+ assertThat(refsTables.history()).hasSize(numSnapshots);
+ }
+
public void testTableAuth(
String catalogToken,
Map<String, String> credentials,
