Skip to content

Feature/182 implement querytable api on icebergsharedtable #214

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.

Sign up for GitHub

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

Merged
merged 9 commits into from
Apr 19, 2024
Merged

Feature/182 implement querytable api on icebergsharedtable #214

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
077b86a
wip
Jan 31, 2024
32d9b6e
add iceberg file stats builder
matar993 Apr 16, 2024
a6e9eae
add IcebergPartitionValuesBuilder
matar993 Apr 17, 2024
4946ff3
Merge branch 'main' into feature/182-implement-querytable-api-on-iceb…
tmnd1991 Apr 18, 2024
3d7e4f5
Piggy back on iceberg to initialize a FileIO
agilelab-tmnd1991 Apr 19, 2024
513e3ee
Disable test on windows and reset coverage
agilelab-tmnd1991 Apr 19, 2024
798420b
Remove TODO
agilelab-tmnd1991 Apr 19, 2024
05f0634
Merge branch 'main' into feature/182-implement-querytable-api-on-iceb…
tmnd1991 Apr 19, 2024
09b07ff
Merge branch 'main' into feature/182-implement-querytable-api-on-iceb…
tmnd1991 Apr 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 117 additions & 2 deletions server/app/src/test/java/io/whitefox/api/deltasharing/SampleTables.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
package io.whitefox.api.deltasharing;

import static io.whitefox.DeltaTestUtils.*;
import static io.whitefox.IcebergTestUtils.icebergTableWithHadoopCatalog;
import static io.whitefox.IcebergTestUtils.s3IcebergTableWithAwsGlueCatalog;
import static io.whitefox.IcebergTestUtils.*;

import io.whitefox.AwsGlueTestConfig;
import io.whitefox.S3TestConfig;
Expand Down Expand Up @@ -73,6 +72,78 @@ public static StorageManager createStorageManager() {
0L)));
}

public static final ParquetProtocol localIcebergTable1Protocol =
ParquetProtocol.ofMinReaderVersion(1);
public static final ParquetProtocol s3IcebergTable1Protocol =
ParquetProtocol.ofMinReaderVersion(1);
public static final ParquetMetadata localIcebergTable1Metadata = ParquetMetadata.builder()
.metadata(ParquetMetadata.Metadata.builder()
.id("3369848726892806393")
.name(Optional.of("metastore.test_db.icebergtable1"))
.format(new Format())
.schemaString(
"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":false,\"metadata\":{}}]}")
.partitionColumns(List.of())
.version(Optional.of(1L))
.configuration(Optional.of(Map.of("write.parquet.compression-codec", "zstd")))
.build())
.build();

public static final Set<FileObjectWithoutPresignedUrl> localIcebergTableFilesToBeSigned = Set.of(
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(419L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":3},\"maxValues\":{\"id\":3},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705667209813L)),
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(419L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":2},\"maxValues\":{\"id\":2},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705667209813L)),
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(419L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":1},\"maxValues\":{\"id\":1},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705667209813L)),
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(418L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":0},\"maxValues\":{\"id\":0},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705667209813L)),
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(419L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":4},\"maxValues\":{\"id\":4},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705667209813L)));

public static final ParquetMetadata s3IcebergTable1Metadata = ParquetMetadata.builder()
.metadata(ParquetMetadata.Metadata.builder()
.id("7819530050735196523")
.name(Optional.of("metastore.test_glue_db.icebergtable1"))
.format(new Format())
.schemaString(
"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":false,\"metadata\":{}}]}")
.partitionColumns(List.of())
.version(Optional.of(1L))
.configuration(Optional.of(Map.of("write.parquet.compression-codec", "zstd")))
.build())
.build();

public static final ParquetMetadata deltaTable1Metadata = ParquetMetadata.builder()
.metadata(ParquetMetadata.Metadata.builder()
.id("56d48189-cdbc-44f2-9b0e-2bded4c79ed7")
Expand All @@ -98,6 +169,7 @@ public static StorageManager createStorageManager() {
.configuration(Optional.of(Map.of()))
.build())
.build();

public static final ParquetMetadata deltaTableWithHistory1Metadata = ParquetMetadata.builder()
.metadata(ParquetMetadata.Metadata.builder()
.id("56d48189-cdbc-44f2-9b0e-2bded4c79ed7")
Expand All @@ -110,6 +182,7 @@ public static StorageManager createStorageManager() {
.configuration(Optional.of(Map.of()))
.build())
.build();

public static final ParquetProtocol deltaTable1Protocol = ParquetProtocol.ofMinReaderVersion(1);

public static final ParquetProtocol s3DeltaTable1Protocol = ParquetProtocol.ofMinReaderVersion(1);
Expand Down Expand Up @@ -196,6 +269,48 @@ public static StorageManager createStorageManager() {
.build())
.build());

public static final Set<FileObjectWithoutPresignedUrl> s3IcebergTable1FilesWithoutPresignedUrl =
Set.of(
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(419L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":4},\"maxValues\":{\"id\":4},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705948389052L)),
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(419L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":3},\"maxValues\":{\"id\":3},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705948389052L)),
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(419L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":2},\"maxValues\":{\"id\":2},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705948389052L)),
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(418L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":0},\"maxValues\":{\"id\":0},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705948389052L)),
new FileObjectWithoutPresignedUrl()
._file(new FileObjectFileWithoutPresignedUrl()
.partitionValues(Map.of())
.size(419L)
.stats(
"{\"numRecords\":1,\"minValues\":{\"id\":1},\"maxValues\":{\"id\":1},\"nullCount\":{\"id\":0}}")
.version(1L)
.timestamp(1705948389052L)));
public static final Set<FileObjectWithoutPresignedUrl> s3DeltaTable1FilesWithoutPresignedUrl =
Set.of(
new FileObjectWithoutPresignedUrl()
Expand Down
4 changes: 4 additions & 0 deletions ...p/src/test/java/io/whitefox/api/deltasharing/model/FileObjectFileWithoutPresignedUrl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,10 @@
import java.util.Map;
import java.util.Objects;

/**
* This class is test-only and it's needed because we can't know in advance pre-signed urls, in this way we can
* easily run assertions on FileObjects that contain pre-signed urls ignoring the url.
*/
public class FileObjectFileWithoutPresignedUrl {

private Map<String, String> partitionValues = new HashMap<>();
Expand Down
4 changes: 4 additions & 0 deletions ...r/app/src/test/java/io/whitefox/api/deltasharing/model/FileObjectWithoutPresignedUrl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,10 @@
import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.Objects;

/**
* This class is test-only and it's needed because we can't know in advance pre-signed urls, in this way we can
* easily run assertions on FileObjects that contain pre-signed urls ignoring the url.
*/
public class FileObjectWithoutPresignedUrl {

private FileObjectFileWithoutPresignedUrl _file;
Expand Down
133 changes: 131 additions & 2 deletions server/app/src/test/java/io/whitefox/api/deltasharing/server/DeltaSharesApiImplAwsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ public void icebergTableMetadata() throws IOException {

@DisabledOnOs(OS.WINDOWS)
@Test
public void queryTableCurrentVersion() throws IOException {
public void queryDeltaTableCurrentVersion() throws IOException {
var responseBodyLines = given()
.when()
.filter(deltaFilter)
Expand Down Expand Up @@ -192,7 +192,7 @@ public void queryTableCurrentVersion() throws IOException {

@DisabledOnOs(OS.WINDOWS)
@Test
public void queryTableByVersion() throws IOException {
public void queryDeltaTableByVersion() throws IOException {
var responseBodyLines = given()
.when()
.filter(deltaFilter)
Expand Down Expand Up @@ -232,4 +232,133 @@ public void queryTableByVersion() throws IOException {
assertEquals(7, responseBodyLines.length);
assertEquals(s3DeltaTable1FilesWithoutPresignedUrl, files);
}

@DisabledOnOs(OS.WINDOWS)
@Test
public void queryIcebergTableCurrentVersion() throws IOException {
var responseBodyLines = given()
.when()
.filter(deltaFilter)
.body("{}")
.header(new Header("Content-Type", "application/json"))
.post(
"delta-api/v1/shares/{share}/schemas/{schema}/tables/{table}/query",
"s3share",
"s3schema",
"s3IcebergTable1")
.then()
.statusCode(200)
.extract()
.body()
.asString()
.split("\n");

assertEquals(
s3IcebergTable1Protocol,
objectMapper.reader().readValue(responseBodyLines[0], ParquetProtocol.class));
assertEquals(
s3IcebergTable1Metadata,
objectMapper.reader().readValue(responseBodyLines[1], ParquetMetadata.class));
var files = Arrays.stream(responseBodyLines)
.skip(2)
.map(line -> {
try {
return objectMapper
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
.reader()
.readValue(line, FileObjectWithoutPresignedUrl.class);
} catch (IOException e) {
throw new RuntimeException(e);
}
})
.collect(Collectors.toSet());
assertEquals(7, responseBodyLines.length);
assertEquals(s3IcebergTable1FilesWithoutPresignedUrl, files);
}

@DisabledOnOs(OS.WINDOWS)
@Test
public void queryIcebergTableByVersion() throws IOException {
var responseBodyLines = given()
.when()
.filter(deltaFilter)
.body("{\"version\": 7819530050735196523}")
.header(new Header("Content-Type", "application/json"))
.post(
"delta-api/v1/shares/{share}/schemas/{schema}/tables/{table}/query",
"s3share",
"s3schema",
"s3IcebergTable1")
.then()
.statusCode(200)
.extract()
.body()
.asString()
.split("\n");

assertEquals(
s3IcebergTable1Protocol,
objectMapper.reader().readValue(responseBodyLines[0], ParquetProtocol.class));
assertEquals(
s3IcebergTable1Metadata,
objectMapper.reader().readValue(responseBodyLines[1], ParquetMetadata.class));
var files = Arrays.stream(responseBodyLines)
.skip(2)
.map(line -> {
try {
return objectMapper
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
.reader()
.readValue(line, FileObjectWithoutPresignedUrl.class);
} catch (IOException e) {
throw new RuntimeException(e);
}
})
.collect(Collectors.toSet());
assertEquals(7, responseBodyLines.length);
assertEquals(s3IcebergTable1FilesWithoutPresignedUrl, files);
}

@DisabledOnOs(OS.WINDOWS)
@Test
public void queryIcebergTableByTs() throws IOException {
var responseBodyLines = given()
.when()
.filter(deltaFilter)
.body("{\"timestamp\": \"2024-02-02T12:00:00Z\"}")
.header(new Header("Content-Type", "application/json"))
.post(
"delta-api/v1/shares/{share}/schemas/{schema}/tables/{table}/query",
"s3share",
"s3schema",
"s3IcebergTable1")
.then()
.statusCode(200)
.extract()
.body()
.asString()
.split("\n");

assertEquals(
s3IcebergTable1Protocol,
objectMapper.reader().readValue(responseBodyLines[0], ParquetProtocol.class));
assertEquals(
s3IcebergTable1Metadata,
objectMapper.reader().readValue(responseBodyLines[1], ParquetMetadata.class));
var files = Arrays.stream(responseBodyLines)
.skip(2)
.map(line -> {
try {
return objectMapper
.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
.reader()
.readValue(line, FileObjectWithoutPresignedUrl.class);
} catch (IOException e) {
throw new RuntimeException(e);
}
})
.collect(Collectors.toSet());
assertEquals(7, responseBodyLines.length);
assertEquals(s3IcebergTable1FilesWithoutPresignedUrl, files);
}
}
Loading