Skip to content

Commit

Permalink
wip
Browse files Browse the repository at this point in the history
  • Loading branch information
@agilelab-tmnd1991
agilelab-tmnd1991 committed Jan 10, 2024
1 parent 7b77e5e commit 7bb1f7e
Show file tree
Hide file tree
Showing 4 changed files with 133 additions and 114 deletions.
59 changes: 34 additions & 25 deletions server/app/src/main/java/io/whitefox/api/deltasharing/DeltaMappers.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,37 +58,46 @@ public static TableQueryResponseObject readTableResult2api(ReadTableResult readT
.collect(Collectors.toList()));
}

private static MetadataObject metadata2Api(Metadata metadata) {
return new MetadataObject()
.metaData(new MetadataObjectMetaData()
.id(metadata.id())
.name(metadata.name().orElse(null))
.description(metadata.description().orElse(null))
.format(new FormatObject().provider(metadata.format().provider()))
.schemaString(metadata.tableSchema().structType().toJson())
.partitionColumns(metadata.partitionColumns())
._configuration(metadata.configuration())
.version(metadata.version().orElse(null))
.numFiles(metadata.numFiles().orElse(null)));
private static ParquetMetadataObject metadata2Api(Metadata metadata) {
return new ParquetMetadataObject()
.metaData(new ParquetMetadataObjectMetaData()
.numFiles(metadata.numFiles().orElse(null))
.version(metadata.version().orElse(null))
.size(metadata.size().orElse(null))
.id(metadata.id())
.name(metadata.name().orElse(null))
.description(metadata.description().orElse(null))
.format(new ParquetFormatObject().provider(metadata.format().provider()))
.schemaString(metadata.tableSchema().structType().toJson())
.partitionColumns(metadata.partitionColumns())
._configuration(metadata.configuration()));
}

private static ProtocolObject protocol2Api(Protocol protocol) {
return new ProtocolObject()
.protocol(new ProtocolObjectProtocol()
.minReaderVersion(protocol.minReaderVersion().orElse(1)));
private static DeltaProtocolObject protocol2Api(Protocol protocol) {
return new DeltaProtocolObject()
.protocol(new DeltaProtocolObjectProtocol()
.deltaProtocol(new DeltaProtocolObjectProtocolDeltaProtocol()
.minReaderVersion(protocol.minReaderVersion().orElse(1))
.minWriterVersion(protocol.minWriterVersion().orElse(1))));
}

private static FileObject file2Api(TableFile f) {
return new FileObject()
._file(new FileObjectFile()
private static DeltaFileObject file2Api(TableFile f) {
return new DeltaFileObject()
.id(f.id())
.url(f.url())
.partitionValues(f.partitionValues())
.size(f.size())
.stats(f.stats().orElse(null))
.version(f.version().orElse(null))
.deletionVectorFileId(null) // TODO
.timestamp(f.timestamp().orElse(null))
.expirationTimestamp(f.expirationTimestamp()));
.expirationTimestamp(f.expirationTimestamp())
.deltaSingleAction(new DeltaSingleAction()
._file(new DeltaAddFileAction()
.id(f.id())
.url(f.url())
.partitionValues(f.partitionValues())
.size(f.size())
.stats(f.stats().orElse(null))
.version(f.version().orElse(null))
.timestamp(f.timestamp().orElse(null))
.expirationTimestamp(f.expirationTimestamp())));
}

public static TableReferenceAndReadRequest api2TableReferenceAndReadRequest(
Expand Down Expand Up @@ -125,7 +134,7 @@ public static Map<String, String> toHeaderCapabilitiesMap(String headerCapabilit

public static TableMetadataResponseObject toTableResponseMetadata(Metadata m) {
return new TableMetadataResponseObject()
.protocol(new ProtocolObject().protocol(new ProtocolObjectProtocol().minReaderVersion(1)))
.protocol(new ParquetProtocolObject().protocol(new ParquetProtocolObjectProtocol().minReaderVersion(1)))
.metadata(metadata2Api(m));
}
}
23 changes: 16 additions & 7 deletions server/core/src/main/java/io/whitefox/core/Protocol.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,19 +6,19 @@

public class Protocol {
private final Optional<Integer> minReaderVersion;
private final Optional<Integer> minWriterVersion;

public Protocol(Optional<Integer> minReaderVersion) {
public Protocol(Optional<Integer> minReaderVersion, Optional<Integer> minWriterVersion) {
this.minReaderVersion = minReaderVersion;
this.minWriterVersion = minWriterVersion;
}

public Optional<Integer> minReaderVersion() {
return minReaderVersion;
}

@Override
@SkipCoverageGenerated
public String toString() {
return "Protocol{" + "minReaderVersion=" + minReaderVersion + '}';
public Optional<Integer> minWriterVersion() {
return minWriterVersion;
}

@Override
Expand All @@ -27,12 +27,21 @@ public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Protocol protocol = (Protocol) o;
return Objects.equals(minReaderVersion, protocol.minReaderVersion);
return Objects.equals(minReaderVersion, protocol.minReaderVersion)
&& Objects.equals(minWriterVersion, protocol.minWriterVersion);
}

@Override
@SkipCoverageGenerated
public int hashCode() {
return Objects.hash(minReaderVersion);
return Objects.hash(minReaderVersion, minWriterVersion);
}

@Override
@SkipCoverageGenerated
public String toString() {
return "Protocol{" + "minReaderVersion="
+ minReaderVersion + ", minWriterVersion="
+ minWriterVersion + '}';
}
}
2 changes: 1 addition & 1 deletion server/core/src/main/java/io/whitefox/core/services/DeltaSharedTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ public ReadTableResultToBeSigned queryTable(ReadTableRequest readTableRequest) {
throw new IllegalArgumentException("Unknown ReadTableRequest type: " + readTableRequest);
}
return new ReadTableResultToBeSigned(
new Protocol(Optional.of(1)),
new Protocol(Optional.of(1), Optional.of(1)),
metadataFromSnapshot(snapshot),
snapshot.getAllFiles().stream()
.map(f -> new TableFileToBeSigned(
Expand Down
163 changes: 82 additions & 81 deletions server/core/src/main/java/io/whitefox/core/services/DeltaSharingCapabilities.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,99 +6,100 @@
import java.util.stream.Collectors;

public class DeltaSharingCapabilities {
private final Map<String, String> values;
private final Map<String, String> values;

public DeltaSharingCapabilities(Map<String, String> values) {
this.values = Map.copyOf(values);
}

public Map<String, String> values() {
return values;
}
public DeltaSharingCapabilities(Map<String, String> values) {
this.values = Map.copyOf(values);
}

public DeltaSharingResponseFormat getResponseFormat() {
var value = values.get(DELTA_SHARING_RESPONSE_FORMAT);
if (value == null) {
return DeltaSharingResponseFormat.PARQUET;
} else {
return DeltaSharingResponseFormat.valueOf(value.toUpperCase());
}
}
public Map<String, String> values() {
return values;
}

public Set<DeltaSharingReaderFeatures> getReaderFeatures() {
var value = values.get(DELTA_SHARING_READER_FEATURES);
if (value == null) {
return Set.of();
} else {
return Arrays.stream(value.split(","))
.map(String::trim)
.map(DeltaSharingReaderFeatures::fromString)
.collect(Collectors.toSet());
}
public DeltaSharingResponseFormat getResponseFormat() {
var value = values.get(DELTA_SHARING_RESPONSE_FORMAT);
if (value == null) {
return DeltaSharingResponseFormat.PARQUET;
} else {
return DeltaSharingResponseFormat.valueOf(value.toUpperCase());
}
}

public final static String DELTA_SHARING_RESPONSE_FORMAT = "responseformat";
public final static String DELTA_SHARING_READER_FEATURES = "readerfeatures";

public enum DeltaSharingResponseFormat {
DELTA, PARQUET
public Set<DeltaSharingReaderFeatures> getReaderFeatures() {
var value = values.get(DELTA_SHARING_READER_FEATURES);
if (value == null) {
return Set.of();
} else {
return Arrays.stream(value.split(","))
.map(String::trim)
.map(DeltaSharingReaderFeatures::fromString)
.collect(Collectors.toSet());
}
}

public enum DeltaSharingReaderFeatures {
DELETION_VECTORS("deletionvectors"),
COLUMN_MAPPING("columnmapping"),
TIMESTAMP_NTZ("timestampntz"),
DOMAIN_METADATA("domainmetadata"),
V2CHECKPOINT("v2checkpoint"),
CHECK_CONSTRAINTS("checkconstraints"),
GENERATED_COLUMNS("generatedcolumns"),
ALLOW_COLUMN_DEFAULTS("allowcolumndefaults"),
IDENTITY_COLUMNS("identitycolumns");
public static final String DELTA_SHARING_RESPONSE_FORMAT = "responseformat";
public static final String DELTA_SHARING_READER_FEATURES = "readerfeatures";

DeltaSharingReaderFeatures(String stringRepresentation) {
this.stringRepresentation = stringRepresentation;
}
public enum DeltaSharingResponseFormat {
DELTA,
PARQUET
}

private final String stringRepresentation;
public enum DeltaSharingReaderFeatures {
DELETION_VECTORS("deletionvectors"),
COLUMN_MAPPING("columnmapping"),
TIMESTAMP_NTZ("timestampntz"),
DOMAIN_METADATA("domainmetadata"),
V2CHECKPOINT("v2checkpoint"),
CHECK_CONSTRAINTS("checkconstraints"),
GENERATED_COLUMNS("generatedcolumns"),
ALLOW_COLUMN_DEFAULTS("allowcolumndefaults"),
IDENTITY_COLUMNS("identitycolumns");

public String stringRepresentation() {
return stringRepresentation;
}
DeltaSharingReaderFeatures(String stringRepresentation) {
this.stringRepresentation = stringRepresentation;
}

public static DeltaSharingReaderFeatures fromString(String s) {
switch (s) {
case DELTA_SHARING_READER_FEATURE_DELETION_VECTOR:
return DeltaSharingReaderFeatures.DELETION_VECTORS;
case DELTA_SHARING_READER_FEATURE_COLUMN_MAPPING:
return DeltaSharingReaderFeatures.COLUMN_MAPPING;
case DELTA_SHARING_READER_FEATURE_TIMESTAMP_NTZ:
return DeltaSharingReaderFeatures.TIMESTAMP_NTZ;
case DELTA_SHARING_READER_FEATURE_DOMAIN_METADATA:
return DeltaSharingReaderFeatures.DOMAIN_METADATA;
case DELTA_SHARING_READER_FEATURE_V2CHECKPOINT:
return DeltaSharingReaderFeatures.V2CHECKPOINT;
case DELTA_SHARING_READER_FEATURE_CHECK_CONSTRAINTS:
return DeltaSharingReaderFeatures.CHECK_CONSTRAINTS;
case DELTA_SHARING_READER_FEATURE_GENERATED_COLUMNS:
return DeltaSharingReaderFeatures.GENERATED_COLUMNS;
case DELTA_SHARING_READER_FEATURE_ALLOW_COLUMN_DEFAULTS:
return DeltaSharingReaderFeatures.ALLOW_COLUMN_DEFAULTS;
case DELTA_SHARING_READER_FEATURE_IDENTITY_COLUMNS:
return DeltaSharingReaderFeatures.IDENTITY_COLUMNS;
default:
throw new IllegalArgumentException("Unknown reader feature: " + s);
}
}
private final String stringRepresentation;

private final static String DELTA_SHARING_READER_FEATURE_DELETION_VECTOR = "deletionvectors";
private final static String DELTA_SHARING_READER_FEATURE_COLUMN_MAPPING = "columnmapping";
private final static String DELTA_SHARING_READER_FEATURE_TIMESTAMP_NTZ = "timestampntz";
private final static String DELTA_SHARING_READER_FEATURE_DOMAIN_METADATA = "domainmetadata";
private final static String DELTA_SHARING_READER_FEATURE_V2CHECKPOINT = "v2checkpoint";
private final static String DELTA_SHARING_READER_FEATURE_CHECK_CONSTRAINTS = "checkconstraints";
private final static String DELTA_SHARING_READER_FEATURE_GENERATED_COLUMNS = "generatedcolumns";
private final static String DELTA_SHARING_READER_FEATURE_ALLOW_COLUMN_DEFAULTS = "allowcolumndefaults";
private final static String DELTA_SHARING_READER_FEATURE_IDENTITY_COLUMNS = "identitycolumns";
public String stringRepresentation() {
return stringRepresentation;
}

public static DeltaSharingReaderFeatures fromString(String s) {
switch (s) {
case DELTA_SHARING_READER_FEATURE_DELETION_VECTOR:
return DeltaSharingReaderFeatures.DELETION_VECTORS;
case DELTA_SHARING_READER_FEATURE_COLUMN_MAPPING:
return DeltaSharingReaderFeatures.COLUMN_MAPPING;
case DELTA_SHARING_READER_FEATURE_TIMESTAMP_NTZ:
return DeltaSharingReaderFeatures.TIMESTAMP_NTZ;
case DELTA_SHARING_READER_FEATURE_DOMAIN_METADATA:
return DeltaSharingReaderFeatures.DOMAIN_METADATA;
case DELTA_SHARING_READER_FEATURE_V2CHECKPOINT:
return DeltaSharingReaderFeatures.V2CHECKPOINT;
case DELTA_SHARING_READER_FEATURE_CHECK_CONSTRAINTS:
return DeltaSharingReaderFeatures.CHECK_CONSTRAINTS;
case DELTA_SHARING_READER_FEATURE_GENERATED_COLUMNS:
return DeltaSharingReaderFeatures.GENERATED_COLUMNS;
case DELTA_SHARING_READER_FEATURE_ALLOW_COLUMN_DEFAULTS:
return DeltaSharingReaderFeatures.ALLOW_COLUMN_DEFAULTS;
case DELTA_SHARING_READER_FEATURE_IDENTITY_COLUMNS:
return DeltaSharingReaderFeatures.IDENTITY_COLUMNS;
default:
throw new IllegalArgumentException("Unknown reader feature: " + s);
}
}

private static final String DELTA_SHARING_READER_FEATURE_DELETION_VECTOR = "deletionvectors";
private static final String DELTA_SHARING_READER_FEATURE_COLUMN_MAPPING = "columnmapping";
private static final String DELTA_SHARING_READER_FEATURE_TIMESTAMP_NTZ = "timestampntz";
private static final String DELTA_SHARING_READER_FEATURE_DOMAIN_METADATA = "domainmetadata";
private static final String DELTA_SHARING_READER_FEATURE_V2CHECKPOINT = "v2checkpoint";
private static final String DELTA_SHARING_READER_FEATURE_CHECK_CONSTRAINTS = "checkconstraints";
private static final String DELTA_SHARING_READER_FEATURE_GENERATED_COLUMNS = "generatedcolumns";
private static final String DELTA_SHARING_READER_FEATURE_ALLOW_COLUMN_DEFAULTS =
"allowcolumndefaults";
private static final String DELTA_SHARING_READER_FEATURE_IDENTITY_COLUMNS = "identitycolumns";
}
}

0 comments on commit 7bb1f7e

Please sign in to comment.