Skip to content
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

[core] Support partition API #4786

Merged
merged 17 commits into from
Dec 30, 2024
Merged
Show file tree
Hide file tree
Changes from 15 commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ public static GenericRow convertSpecToInternalRow(
List<String> fieldNames = partType.getFieldNames();
for (Map.Entry<String, String> entry : spec.entrySet()) {
Object value =
defaultPartValue.equals(entry.getValue())
defaultPartValue != null && defaultPartValue.equals(entry.getValue())
? null
: castFromString(
entry.getValue(), partType.getField(entry.getKey()).type());
Expand Down
17 changes: 17 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/rest/HttpClient.java
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,23 @@ public <T extends RESTResponse> T delete(String path, Map<String, String> header
return exec(request, null);
}

@Override
public <T extends RESTResponse> T delete(
String path, RESTRequest body, Map<String, String> headers) {
try {
RequestBody requestBody = buildRequestBody(body);
Request request =
new Request.Builder()
.url(uri + path)
.delete(requestBody)
.headers(Headers.of(headers))
.build();
return exec(request, null);
} catch (JsonProcessingException e) {
throw new RESTException(e, "build request failed.");
}
}

@Override
public void close() throws IOException {
okHttpClient.dispatcher().cancelAll();
Expand Down
117 changes: 112 additions & 5 deletions paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,19 @@

import org.apache.paimon.CoreOptions;
import org.apache.paimon.TableType;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogUtils;
import org.apache.paimon.catalog.Database;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.catalog.PropertyChange;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.operation.FileStoreCommit;
import org.apache.paimon.operation.Lock;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
Expand All @@ -41,15 +45,19 @@
import org.apache.paimon.rest.requests.AlterDatabaseRequest;
import org.apache.paimon.rest.requests.AlterTableRequest;
import org.apache.paimon.rest.requests.CreateDatabaseRequest;
import org.apache.paimon.rest.requests.CreatePartitionRequest;
import org.apache.paimon.rest.requests.CreateTableRequest;
import org.apache.paimon.rest.requests.DropPartitionRequest;
import org.apache.paimon.rest.requests.RenameTableRequest;
import org.apache.paimon.rest.responses.AlterDatabaseResponse;
import org.apache.paimon.rest.responses.ConfigResponse;
import org.apache.paimon.rest.responses.CreateDatabaseResponse;
import org.apache.paimon.rest.responses.GetDatabaseResponse;
import org.apache.paimon.rest.responses.GetTableResponse;
import org.apache.paimon.rest.responses.ListDatabasesResponse;
import org.apache.paimon.rest.responses.ListPartitionsResponse;
import org.apache.paimon.rest.responses.ListTablesResponse;
import org.apache.paimon.rest.responses.PartitionResponse;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.TableSchema;
Expand All @@ -58,10 +66,11 @@
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.object.ObjectTable;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.Preconditions;

import org.apache.paimon.shade.guava30.com.google.common.annotations.VisibleForTesting;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;

Expand All @@ -71,15 +80,20 @@
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;

import static org.apache.paimon.CoreOptions.createCommitUser;
import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemDatabase;
import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemTable;
import static org.apache.paimon.catalog.CatalogUtils.isSystemDatabase;
import static org.apache.paimon.options.CatalogOptions.CASE_SENSITIVE;
import static org.apache.paimon.utils.InternalRowPartitionComputer.convertSpecToInternalRow;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
import static org.apache.paimon.utils.ThreadPoolUtils.createScheduledThreadPool;

Expand Down Expand Up @@ -132,7 +146,8 @@ public RESTCatalog(CatalogContext catalogContext) {
Map<String, String> initHeaders =
RESTUtil.merge(
configHeaders(catalogOptions.toMap()), this.catalogAuth.getHeaders());
Options options = new Options(fetchOptionsFromServer(initHeaders, initHeaders));
Options options =
new Options(fetchOptionsFromServer(initHeaders, catalogContext.options().toMap()));
this.context =
CatalogContext.create(
options, catalogContext.preferIO(), catalogContext.fallbackIO());
Expand Down Expand Up @@ -360,17 +375,43 @@ public void dropTable(Identifier identifier, boolean ignoreIfNotExists)
@Override
public void createPartition(Identifier identifier, Map<String, String> partitionSpec)
throws TableNotExistException {
throw new UnsupportedOperationException();
try {
CreatePartitionRequest request = new CreatePartitionRequest(identifier, partitionSpec);
client.post(
resourcePaths.partitions(
identifier.getDatabaseName(), identifier.getTableName()),
request,
PartitionResponse.class,
headers());
} catch (NoSuchResourceException e) {
throw new TableNotExistException(identifier);
} catch (ForbiddenException e) {
throw new TableNoPermissionException(identifier, e);
}
}

@Override
public void dropPartition(Identifier identifier, Map<String, String> partitions)
throws TableNotExistException, PartitionNotExistException {}
throws TableNotExistException, PartitionNotExistException {
checkNotSystemTable(identifier, "dropPartition");
dropPartitionMetadata(identifier, partitions);
Table table = getTable(identifier);
cleanPartitionsInFileSystem(table, partitions);
}

@Override
public List<PartitionEntry> listPartitions(Identifier identifier)
throws TableNotExistException {
throw new UnsupportedOperationException();
FileStoreTable table = (FileStoreTable) getTable(identifier);
boolean whetherSupportListPartitions =
Boolean.parseBoolean(
table.options().get(CoreOptions.METASTORE_PARTITIONED_TABLE.key()));
if (whetherSupportListPartitions) {
RowType rowType = table.schema().logicalPartitionType();
return listPartitionsFromServer(identifier, rowType);
} else {
return getTable(identifier).newReadBuilder().newScan().listPartitionEntries();
}
}

@Override
Expand Down Expand Up @@ -420,6 +461,55 @@ Table getDataOrFormatTable(Identifier identifier) throws TableNotExistException
return table;
}

@VisibleForTesting
public List<PartitionEntry> listPartitionsFromServer(Identifier identifier, RowType rowType)
throws TableNotExistException {
try {
ListPartitionsResponse response =
client.get(
resourcePaths.partitions(
identifier.getDatabaseName(), identifier.getTableName()),
ListPartitionsResponse.class,
headers());
if (response != null && response.getPartitions() != null) {
return response.getPartitions().stream()
.map(p -> convertToPartitionEntry(p, rowType))
.collect(Collectors.toList());
} else {
return Collections.emptyList();
}
} catch (NoSuchResourceException e) {
throw new TableNotExistException(identifier);
} catch (ForbiddenException e) {
throw new TableNoPermissionException(identifier, e);
}
}

@VisibleForTesting
PartitionEntry convertToPartitionEntry(PartitionResponse partition, RowType rowType) {
InternalRowSerializer serializer = new InternalRowSerializer(rowType);
GenericRow row = convertSpecToInternalRow(partition.getSpec(), rowType, null);
return new PartitionEntry(
serializer.toBinaryRow(row).copy(),
partition.getRecordCount(),
partition.getFileSizeInBytes(),
partition.getFileCount(),
partition.getLastFileCreationTime());
}

@VisibleForTesting
void cleanPartitionsInFileSystem(Table table, Map<String, String> partitions) {
FileStoreTable fileStoreTable = (FileStoreTable) table;
try (FileStoreCommit commit =
fileStoreTable
.store()
.newCommit(
createCommitUser(fileStoreTable.coreOptions().toConfiguration()))) {
commit.dropPartitions(
Collections.singletonList(partitions), BatchWriteBuilder.COMMIT_IDENTIFIER);
}
}

protected GetTableResponse getTableResponse(Identifier identifier)
throws TableNotExistException {
try {
Expand All @@ -434,6 +524,23 @@ protected GetTableResponse getTableResponse(Identifier identifier)
}
}

protected boolean dropPartitionMetadata(Identifier identifier, Map<String, String> partitions)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why need these tests invoking this method? Why not invoking catalog methods?

throws TableNoPermissionException {
try {
DropPartitionRequest request = new DropPartitionRequest(partitions);
client.delete(
resourcePaths.partitions(
identifier.getDatabaseName(), identifier.getTableName()),
request,
headers());
return true;
} catch (NoSuchResourceException ignore) {
return true;
} catch (ForbiddenException e) {
throw new TableNoPermissionException(identifier, e);
}
}

private static Map<String, String> configHeaders(Map<String, String> properties) {
return RESTUtil.extractPrefixMap(properties, "header.");
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,4 +30,6 @@ <T extends RESTResponse> T post(
String path, RESTRequest body, Class<T> responseType, Map<String, String> headers);

<T extends RESTResponse> T delete(String path, Map<String, String> headers);

<T extends RESTResponse> T delete(String path, RESTRequest body, Map<String, String> headers);
}
Original file line number Diff line number Diff line change
Expand Up @@ -82,4 +82,15 @@ public String renameTable(String databaseName, String tableName) {
.add("rename")
.toString();
}

public String partitions(String databaseName, String tableName) {
return SLASH.add("v1")
.add(prefix)
.add("databases")
.add(databaseName)
.add("tables")
.add(tableName)
.add("partitions")
.toString();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
/*
* 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.paimon.rest.requests;

import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.rest.RESTRequest;

import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;

import java.util.Map;

/** Request for creating partition. */
@JsonIgnoreProperties(ignoreUnknown = true)
public class CreatePartitionRequest implements RESTRequest {

private static final String FIELD_IDENTIFIER = "identifier";
private static final String FIELD_PARTITION_SPEC = "spec";

@JsonProperty(FIELD_IDENTIFIER)
private final Identifier identifier;

@JsonProperty(FIELD_PARTITION_SPEC)
private final Map<String, String> partitionSpec;

@JsonCreator
public CreatePartitionRequest(
@JsonProperty(FIELD_IDENTIFIER) Identifier identifier,
@JsonProperty(FIELD_PARTITION_SPEC) Map<String, String> partitionSpec) {
this.identifier = identifier;
this.partitionSpec = partitionSpec;
}

@JsonGetter(FIELD_IDENTIFIER)
public Identifier getIdentifier() {
return identifier;
}

@JsonGetter(FIELD_PARTITION_SPEC)
public Map<String, String> getPartitionSpec() {
return partitionSpec;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
/*
* 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.paimon.rest.requests;

import org.apache.paimon.rest.RESTRequest;

import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;

import java.util.Map;

/** Request for deleting partition. */
@JsonIgnoreProperties(ignoreUnknown = true)
public class DropPartitionRequest implements RESTRequest {

private static final String FIELD_PARTITION_SPEC = "spec";

@JsonProperty(FIELD_PARTITION_SPEC)
private final Map<String, String> partitionSpec;

@JsonCreator
public DropPartitionRequest(
@JsonProperty(FIELD_PARTITION_SPEC) Map<String, String> partitionSpec) {
this.partitionSpec = partitionSpec;
}

@JsonGetter(FIELD_PARTITION_SPEC)
public Map<String, String> getPartitionSpec() {
return partitionSpec;
}
}
Loading
Loading