Skip to content

Commit

Permalink
add ut
Browse files Browse the repository at this point in the history
  • Loading branch information
@jerry-024
jerry-024 committed Dec 24, 2024
1 parent 00a0637 commit 43f2353
Show file tree
Hide file tree
Showing 5 changed files with 281 additions and 88 deletions.
96 changes: 50 additions & 46 deletions paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -299,7 +299,9 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx
GetTableResponse.class,
headers());
} catch (AlreadyExistsException e) {
throw new TableAlreadyExistException(identifier);
if (!ignoreIfExists) {
throw new TableAlreadyExistException(identifier);
}
}
}

Expand Down Expand Up @@ -389,12 +391,49 @@ Map<String, String> fetchOptionsFromServer(
return response.merge(clientProperties);
}

private static Map<String, String> configHeaders(Map<String, String> properties) {
return RESTUtil.extractPrefixMap(properties, "header.");
// todo: how know which exception to throw
@VisibleForTesting
void updateTable(Identifier fromTable, Identifier toTable, List<SchemaChange> changes) {
UpdateTableRequest request =
new UpdateTableRequest(fromTable, toTable, new SchemaChanges(changes));
client.post(
resourcePaths.table(fromTable.getDatabaseName(), fromTable.getTableName()),
request,
GetTableResponse.class,
headers());
}

private Map<String, String> headers() {
return catalogAuth.getHeaders();
@VisibleForTesting
Table getDataOrFormatTable(Identifier identifier) throws TableNotExistException {
Preconditions.checkArgument(identifier.getSystemTableName() == null);
TableSchema tableSchema = getDataTableSchema(identifier);
String uuid = null;
FileStoreTable table =
FileStoreTableFactory.create(
fileIO,
newTableLocation(warehouse(), identifier),
tableSchema,
new CatalogEnvironment(
identifier,
uuid,
Lock.factory(
lockFactory(context.options(), fileIO, Optional.empty())
.orElse(null),
lockContext(context.options()).orElse(null),
identifier),
null)); // todo: whether need MetastoreClient.Factory
CoreOptions options = table.coreOptions();
if (options.type() == TableType.OBJECT_TABLE) {
String objectLocation = options.objectLocation();
checkNotNull(objectLocation, "Object location should not be null for object table.");
table =
ObjectTable.builder()
.underlyingTable(table)
.objectLocation(objectLocation)
.objectFileIO(this.fileIO)
.build();
}
return table;
}

protected TableSchema getDataTableSchema(Identifier identifier) throws TableNotExistException {
Expand All @@ -413,15 +452,12 @@ protected TableSchema getDataTableSchema(Identifier identifier) throws TableNotE
}
}

// todo: how know which exception to throw
private void updateTable(Identifier fromTable, Identifier toTable, List<SchemaChange> changes) {
UpdateTableRequest request =
new UpdateTableRequest(fromTable, toTable, new SchemaChanges(changes));
client.post(
resourcePaths.table(fromTable.getDatabaseName(), fromTable.getTableName()),
request,
GetTableResponse.class,
headers());
private static Map<String, String> configHeaders(Map<String, String> properties) {
return RESTUtil.extractPrefixMap(properties, "header.");
}

private Map<String, String> headers() {
return catalogAuth.getHeaders();
}

private Table getAllInSystemDatabase(Identifier identifier) throws TableNotExistException {
Expand Down Expand Up @@ -465,38 +501,6 @@ private Table getSystemTable(Identifier identifier) throws TableNotExistExceptio
return CatalogUtils.getSystemTable(identifier, originTable);
}

private Table getDataOrFormatTable(Identifier identifier) throws TableNotExistException {
Preconditions.checkArgument(identifier.getSystemTableName() == null);
TableSchema tableSchema = getDataTableSchema(identifier);
String uuid = null;
FileStoreTable table =
FileStoreTableFactory.create(
fileIO,
newTableLocation(warehouse(), identifier),
tableSchema,
new CatalogEnvironment(
identifier,
uuid,
Lock.factory(
lockFactory(context.options(), fileIO, Optional.empty())
.orElse(null),
lockContext(context.options()).orElse(null),
identifier),
null)); // todo: whether need MetastoreClient.Factory
CoreOptions options = table.coreOptions();
if (options.type() == TableType.OBJECT_TABLE) {
String objectLocation = options.objectLocation();
checkNotNull(objectLocation, "Object location should not be null for object table.");
table =
ObjectTable.builder()
.underlyingTable(table)
.objectLocation(objectLocation)
.objectFileIO(this.fileIO)
.build();
}
return table;
}

private ScheduledExecutorService tokenRefreshExecutor() {
if (refreshExecutor == null) {
synchronized (this) {
Expand Down
7 changes: 7 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/rest/RESTObjectMapper.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@

package org.apache.paimon.rest;

import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.schema.TableSchemaSerializer;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypeJsonParser;
Expand All @@ -44,6 +46,11 @@ public static ObjectMapper create() {

public static Module createPaimonRestJacksonModule() {
SimpleModule module = new SimpleModule("Paimon_REST");
registerJsonObjects(
module,
TableSchema.class,
TableSchemaSerializer.INSTANCE,
TableSchemaSerializer.INSTANCE);
registerJsonObjects(
module,
DataField.class,
Expand Down
112 changes: 70 additions & 42 deletions paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package org.apache.paimon.rest;

import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.rest.requests.AlterDatabaseRequest;
import org.apache.paimon.rest.requests.CreateDatabaseRequest;
Expand All @@ -28,17 +29,23 @@
import org.apache.paimon.rest.responses.CreateDatabaseResponse;
import org.apache.paimon.rest.responses.ErrorResponse;
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.ListTablesResponse;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;

import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -120,13 +127,21 @@ public static CreateTableRequest createTableRequest(String name) {
public static UpdateTableRequest updateTableRequest(String fromTableName, String toTableName) {
Identifier fromIdentifier = Identifier.create(databaseName(), fromTableName);
Identifier toIdentifier = Identifier.create(databaseName(), toTableName);
SchemaChanges changes = new SchemaChanges(getChanges());
return new UpdateTableRequest(fromIdentifier, toIdentifier, changes);
}

public static List<SchemaChange> getChanges() {
// add option
SchemaChange addOption = SchemaChange.setOption("snapshot.time-retained", "2h");
// remove option
SchemaChange removeOption = SchemaChange.removeOption("compaction.max.file-num");
// add column
SchemaChange addColumn =
SchemaChange.addColumn("col1_after", DataTypes.ARRAY(DataTypes.STRING()));
SchemaChange addColumnMap =
SchemaChange.addColumn(
"col1_map_type", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()));
RowType rowType =
RowType.of(
new DataType[] {
Expand All @@ -139,54 +154,67 @@ public static UpdateTableRequest updateTableRequest(String fromTableName, String
DataTypes.MULTISET(DataTypes.VARCHAR(8))
},
new String[] {"pt", "a", "b", "c", "d", "e", "f"});
SchemaChange addColumnMap =
SchemaChange.addColumn(
"col11_map_type", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()));
SchemaChange addColumnRowType = SchemaChange.addColumn("col11_row_type", rowType);
// // add a column after col1
// SchemaChange.Move after = SchemaChange.Move.after("col1_after", "col1");
// SchemaChange addColumnAfterField =
// SchemaChange.addColumn("col7", DataTypes.STRING(), "", after);
// // rename column
// SchemaChange renameColumn = SchemaChange.renameColumn("col3", "col3_new_name");
// // drop column
// SchemaChange dropColumn = SchemaChange.dropColumn("col6");
// // update column comment
// SchemaChange updateColumnComment =
// SchemaChange.updateColumnComment(new String[] {"col4"}, "col4 field");
// // update nested column comment
// SchemaChange updateNestedColumnComment =
// SchemaChange.updateColumnComment(new String[] {"col5", "f1"}, "col5 f1
// field");
// // update column type
// SchemaChange updateColumnType = SchemaChange.updateColumnType("col4",
// DataTypes.DOUBLE());
// // update column position, you need to pass in a parameter of type Move
// SchemaChange updateColumnPosition =
// SchemaChange.updateColumnPosition(SchemaChange.Move.first("col4"));
// // update column nullability
// SchemaChange updateColumnNullability =
// SchemaChange.updateColumnNullability(new String[] {"col4"}, false);
// // update nested column nullability
// SchemaChange updateNestedColumnNullability =
// SchemaChange.updateColumnNullability(new String[] {"col5", "f2"}, false);
SchemaChange addColumnRowType = SchemaChange.addColumn("col_row_type", rowType);
// add a column after col1
SchemaChange.Move after = SchemaChange.Move.after("col1_after", "col1");
SchemaChange addColumnAfterField =
SchemaChange.addColumn("col7", DataTypes.STRING(), "", after);
// rename column
SchemaChange renameColumn = SchemaChange.renameColumn("col3", "col3_new_name");
// drop column
SchemaChange dropColumn = SchemaChange.dropColumn("col6");
// update column comment
SchemaChange updateColumnComment =
SchemaChange.updateColumnComment(new String[] {"col4"}, "col4 field");
// update nested column comment
SchemaChange updateNestedColumnComment =
SchemaChange.updateColumnComment(new String[] {"col5", "f1"}, "col5 f1 field");
// update column type
SchemaChange updateColumnType = SchemaChange.updateColumnType("col4", DataTypes.DOUBLE());
// update column position, you need to pass in a parameter of type Move
SchemaChange updateColumnPosition =
SchemaChange.updateColumnPosition(SchemaChange.Move.first("col4"));
// update column nullability
SchemaChange updateColumnNullability =
SchemaChange.updateColumnNullability(new String[] {"col4"}, false);
// update nested column nullability
SchemaChange updateNestedColumnNullability =
SchemaChange.updateColumnNullability(new String[] {"col5", "f2"}, false);

List<SchemaChange> schemaChanges = new ArrayList<>();
schemaChanges.add(addOption);
schemaChanges.add(removeOption);
schemaChanges.add(addColumn);
schemaChanges.add(addColumnMap);
schemaChanges.add(addColumnRowType);
// schemaChanges.add(addColumnAfterField);
// schemaChanges.add(renameColumn);
// schemaChanges.add(dropColumn);
// schemaChanges.add(updateColumnComment);
// schemaChanges.add(updateNestedColumnComment);
// schemaChanges.add(updateColumnType);
// schemaChanges.add(updateColumnPosition);
// schemaChanges.add(updateColumnNullability);
// schemaChanges.add(updateNestedColumnNullability);
SchemaChanges changes = new SchemaChanges(schemaChanges);
return new UpdateTableRequest(fromIdentifier, toIdentifier, changes);
schemaChanges.add(addColumnAfterField);
schemaChanges.add(renameColumn);
schemaChanges.add(dropColumn);
schemaChanges.add(updateColumnComment);
schemaChanges.add(updateNestedColumnComment);
schemaChanges.add(updateColumnType);
schemaChanges.add(updateColumnPosition);
schemaChanges.add(updateColumnNullability);
schemaChanges.add(updateNestedColumnNullability);
return schemaChanges;
}

public static GetTableResponse getTableResponse() {
return new GetTableResponse("location", tableSchema());
}

private static TableSchema tableSchema() {
List<DataField> fields =
Arrays.asList(
new DataField(0, "f0", new IntType()),
new DataField(1, "f1", new IntType()));
List<String> partitionKeys = Collections.singletonList("f0");
List<String> primaryKeys = Arrays.asList("f0", "f1");
Map<String, String> options = new HashMap<>();
options.put("option-1", "value-1");
options.put("option-2", "value-2");
// set path for test as if not set system will add one
options.put(CoreOptions.PATH.key(), "/a/b/c");
return new TableSchema(1, fields, 1, partitionKeys, primaryKeys, options, "comment");
}
}
Loading

0 comments on commit 43f2353

Please sign in to comment.