Merge branch 'main' into 184-query-the-iceberg-table-through-all-delta-tablev1-apis

Author: agilelab-tmnd1991

server/app/src/test/java/io/whitefox/api/deltasharing/SampleTables.java

@@ -41,6 +41,9 @@ public static InternalTable s3IcebergTable1(
   public static final InternalTable icebergtable1 =
       icebergTableWithHadoopCatalog("test_db", "icebergtable1");
 
+  public static final InternalTable icebergtable2 =
+      icebergTableWithHadoopCatalog("test_db", "icebergtable2");
+
   public static final String deltaTable1Path = deltaTableUri("delta-table");
 
   public static final String deltaTableWithHistory1Path = deltaTableUri("delta-table-with-history");
@@ -59,7 +62,8 @@ public static StorageManager createStorageManager() {
                     new SharedTable("table1", "default", "name", deltaTable1),
                     new SharedTable(
                         "table-with-history", "default", "name", deltaTableWithHistory1),
-                    new SharedTable("icebergtable1", "default", "name", icebergtable1)),
+                    new SharedTable("icebergtable1", "default", "name", icebergtable1),
+                    new SharedTable("icebergtable2", "default", "name", icebergtable2)),
                 "name")),
         testPrincipal,
         0L)));

server/app/src/test/java/io/whitefox/api/deltasharing/server/DeltaSharesApiImplAwsTest.java

@@ -88,6 +88,22 @@ public void updateStorageManagerWithS3Tables() {
         0L));
   }
 
+  @Test
+  @DisabledOnOs(OS.WINDOWS)
+  public void icebergTableVersion() {
+    given()
+        .when()
+        .filter(deltaFilter)
+        .get(
+            "delta-api/v1/shares/{share}/schemas/{schema}/tables/{table}/version",
+            "s3share",
+            "s3schema",
+            "s3IcebergTable1")
+        .then()
+        .statusCode(200)
+        .header("Delta-Table-Version", "1");
+  }
+
   @Test
   @DisabledOnOs(OS.WINDOWS)
   public void icebergTableMetadata() throws IOException {

server/app/src/test/java/io/whitefox/api/deltasharing/server/DeltaSharesApiImplTest.java

@@ -149,10 +149,13 @@ public void listTables() {
         .get("delta-api/v1/shares/{share}/schemas/{schema}/tables", "name", "default")
         .then()
         .statusCode(200)
-        .body("items", hasSize(3))
+        .body("items", hasSize(4))
         .body(
             "items[0].name",
-            either(is("table1")).or(is("table-with-history")).or(is("icebergtable1")))
+            either(is("table1"))
+                .or(is("table-with-history"))
+                .or(is("icebergtable1"))
+                .or(is("icebergtable2")))
         .body("items[0].schema", is("default"))
         .body("items[0].share", is("name"))
         .body("nextPageToken", is(nullValue()));
@@ -206,6 +209,22 @@ public void deltaTableMetadata() throws IOException {
         objectMapper.reader().readValue(responseBodyLines[1], MetadataObject.class));
   }
 
+  @Test
+  @DisabledOnOs(OS.WINDOWS)
+  public void icebergTableVersion() {
+    given()
+        .when()
+        .filter(deltaFilter)
+        .get(
+            "delta-api/v1/shares/{share}/schemas/{schema}/tables/{table}/version",
+            "name",
+            "default",
+            "icebergtable1")
+        .then()
+        .statusCode(200)
+        .header("Delta-Table-Version", "1");
+  }
+
   @Test
   @DisabledOnOs(OS.WINDOWS)
   public void icebergTableMetadata() throws IOException {
@@ -248,10 +267,13 @@ public void listAllTables() {
         .get("delta-api/v1/shares/{share}/all-tables", "name")
         .then()
         .statusCode(200)
-        .body("items", hasSize(3))
+        .body("items", hasSize(4))
         .body(
             "items[0].name",
-            either(is("table1")).or(is("table-with-history")).or(is("icebergtable1")))
+            either(is("table1"))
+                .or(is("table-with-history"))
+                .or(is("icebergtable1"))
+                .or(is("icebergtable2")))
         .body("items[0].schema", is("default"))
         .body("items[0].share", is("name"))
         .body("nextPageToken", is(nullValue()));

server/core/src/main/java/io/whitefox/core/services/IcebergSharedTable.java

@@ -13,6 +13,7 @@
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.util.SnapshotUtil;
 
 public class IcebergSharedTable implements InternalSharedTable {
 
@@ -59,8 +60,17 @@ private Optional<Snapshot> getSnapshot(Optional<String> startingTimestamp) {
     return startingTimestamp
         .map(this::getTimestamp)
         .map(Timestamp::getTime)
-        .map(icebergTable::snapshot)
-        .or(() -> Optional.of(icebergTable.currentSnapshot()));
+        .map(this::getSnapshotForTimestampAsOf)
+        .orElseGet(() -> Optional.ofNullable(icebergTable.currentSnapshot()));
+  }
+
+  private Optional<Snapshot> getSnapshotForTimestampAsOf(long l) {
+    try {
+      return Optional.of(SnapshotUtil.snapshotIdAsOfTime(icebergTable, l))
+          .map(icebergTable::snapshot);
+    } catch (IllegalArgumentException iea) {
+      return Optional.empty();
+    }
   }
 
   private Timestamp getTimestamp(String timestamp) {
@@ -71,7 +81,7 @@ private Timestamp getTimestamp(String timestamp) {
 
   @Override
   public Optional<Long> getTableVersion(Optional<String> startingTimestamp) {
-    return Optional.of(0L);
+    return getSnapshot(startingTimestamp).map(Snapshot::sequenceNumber);
   }
 
   @Override

server/core/src/test/java/io/whitefox/core/services/IcebergSharedTableTest.java

@@ -30,10 +30,61 @@ void getTableMetadata() {
     assertEquals("3369848726892806393", metadata.get().id());
   }
 
+  @Test
+  void getTableMetadataWithTimestamp() {
+    var PTable = new SharedTable(
+        "icebergtable2",
+        "default",
+        "share1",
+        icebergTableWithHadoopCatalog("test_db", "icebergtable2"));
+    var DTable = icebergTableLoader.loadTable(PTable);
+    var metadata = DTable.getMetadata(Optional.of("2024-01-25T01:32:15+01:00"));
+    assertTrue(metadata.isPresent());
+    assertEquals("2174306913745765008", metadata.get().id());
+  }
+
   @Test
   void getUnknownTableMetadata() {
     var unknownPTable = new SharedTable(
         "notFound", "default", "share1", icebergTableWithHadoopCatalog("test_db", "not-found"));
     assertThrows(IllegalArgumentException.class, () -> DeltaSharedTable.of(unknownPTable));
   }
+
+  @Test
+  void getTableVersion() {
+    var PTable = new SharedTable(
+        "icebergtable1",
+        "default",
+        "share1",
+        icebergTableWithHadoopCatalog("test_db", "icebergtable1"));
+    var DTable = icebergTableLoader.loadTable(PTable);
+    var version = DTable.getTableVersion(Optional.empty());
+    assertTrue(version.isPresent());
+    assertEquals(1, version.get());
+  }
+
+  @Test
+  void getTableVersionWithTimestamp() {
+    var PTable = new SharedTable(
+        "icebergtable2",
+        "default",
+        "share1",
+        icebergTableWithHadoopCatalog("test_db", "icebergtable2"));
+    var DTable = icebergTableLoader.loadTable(PTable);
+    var version = DTable.getTableVersion(Optional.of("2024-01-25T01:32:15+01:00"));
+    assertTrue(version.isPresent());
+    assertEquals(1, version.get());
+  }
+
+  @Test
+  void getTableVersionWithTooOldTimestamp() {
+    var PTable = new SharedTable(
+        "icebergtable2",
+        "default",
+        "share1",
+        icebergTableWithHadoopCatalog("test_db", "icebergtable2"));
+    var DTable = icebergTableLoader.loadTable(PTable);
+    var version = DTable.getTableVersion(Optional.of("2024-01-24T01:32:15+01:00"));
+    assertTrue(version.isEmpty());
+  }
 }