You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/08/27 05:40:33 UTC
[GitHub] [iceberg] JingsongLi opened a new pull request #1393: Flink: Support creating table and altering table in Flink SQL
JingsongLi opened a new pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393
Fixes #1392
1. Create table: Only support identity partition, wait for Flink SQL support hidden partition, but can load all Iceberg tables.
2. Alter table:
- Currently, Flink SQL only support altering table properties, not support altering schema and partition.
- Also support `location`, `current-snapshot-id` and `cherry-pick-snapshot-id` like Spark.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#issuecomment-686058432
+1
Thank @JingsongLi, I merged this to master. And thanks to @openinx and @kbendick for reviewing!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#issuecomment-683516342
Thanks a lot for your review~ I'll take a look~
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479567345
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
+ }
+ }
+ return partitionKeys;
+ }
+
+ private static void commitChanges(Table table, String setLocation, String setSnapshotId,
+ String pickSnapshotId, Map<String, String> setProperties) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is ambiguous and choosing
+ // one order leads to different results
+ Preconditions.checkArgument(setSnapshotId == null || pickSnapshotId == null,
+ "Cannot set the current the current snapshot ID and cherry-pick snapshot changes");
+
+ if (setSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(setSnapshotId);
+ table.manageSnapshots().setCurrentSnapshot(newSnapshotId).commit();
+ }
+
+ // if updating the table snapshot, perform that update first in case it fails
+ if (pickSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(pickSnapshotId);
+ table.manageSnapshots().cherrypick(newSnapshotId).commit();
+ }
+
+ Transaction transaction = table.newTransaction();
Review comment:
Should we do the operations above this point in the transaction as well? That seems reasonable to me. I'm not sure why we don't in other places.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479565256
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
Review comment:
I should also note that support for adding/removing/renaming columns cannot be done by comparing `CatalogTable` instances, unless the Flink schema contains Iceberg column IDs.
The problem is clear when you consider a simple example:
* Iceberg table schema: `id bigint, a float, b float`
* Flink table schema: `id bigint, x float, y float`
There are two ways to get the Flink schema: rename a -> x and b -> y, or drop a, drop b, add x, add y. Guessing which one was intended by the user is not okay because it would corrupt data. If the values from a are read when projecting x after a was actually dropped, then this is a serious correctness bug.
Also note that there are some transformations that can't be detected. For example, drop a then add a. The result should be that all values of column a are discarded. This happens when the wrong data was written to a column but the column is still needed for newer data.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478318403
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +286,29 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = getIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table getIcebergTable(ObjectPath tablePath) throws TableNotExistException {
+ try {
+ return icebergCatalog.loadTable(toIdentifier(tablePath));
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
throw new TableNotExistException(getName(), tablePath, e);
}
}
+ private CatalogTable toCatalogTable(Table table) {
+ TableSchema schema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema());
+
+ // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
Review comment:
I mean, we cannot create a `IcebergCatalogTable` class extends `CatalogTable` to carry iceberg `Table` Object.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479562811
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +287,29 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = getIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table getIcebergTable(ObjectPath tablePath) throws TableNotExistException {
+ try {
+ return icebergCatalog.loadTable(toIdentifier(tablePath));
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
throw new TableNotExistException(getName(), tablePath, e);
}
}
+ private CatalogTable toCatalogTable(Table table) {
+ TableSchema schema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema());
+
+ // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
+ // catalog table.
+ // Let's re-loading table from Iceberg catalog when creating source/sink operators.
+ return new CatalogTableImpl(schema, partitionKeys, table.properties(), null);
Review comment:
What is null? Could you add a comment?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue merged pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue merged pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479869687
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
Review comment:
I think there is no conflict/error, but I think it is good to reduce duplicate storage, cause iceberg has saved this information.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479712374
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -88,4 +105,172 @@ public void testRenameTable() {
Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
tEnv.from("tl2").getSchema().getTableColumns());
}
+
+ @Test
+ public void testCreateTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ }
+
+ @Test
+ public void testCreateTableLocation() {
+ Assume.assumeFalse("HadoopCatalog does not support creating table with location", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('location'='/tmp/location')");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals("/tmp/location", table.location());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+ }
+
+ @Test
+ public void testCreatePartitionTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT, dt STRING) PARTITIONED BY(dt)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(PartitionSpec.builderFor(table.schema()).identity("dt").build(), table.spec());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).field("dt", DataTypes.STRING()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.singletonList("dt"), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testLoadTransformPartitionTable() throws TableNotExistException {
+ Schema schema = new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
+ validationCatalog.createTable(
+ TableIdentifier.of(icebergNamespace, "tl"), schema,
+ PartitionSpec.builderFor(schema).bucket("id", 100).build());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.emptyList(), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testAlterTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ tEnv.executeSql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ tEnv.executeSql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ CatalogTable catalogTable = catalogTable("tl");
+ properties.remove("oldK");
+ tEnv.getCatalog(tEnv.getCurrentCatalog()).get().alterTable(
+ new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testRelocateTable() {
+ Assume.assumeFalse("HadoopCatalog does not support relocate table", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+ tEnv.executeSql("ALTER TABLE tl SET('location'='/tmp/location')");
Review comment:
According to this documentation, the `LOCATION` clause is supported in a create table statement.
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/create.html#create-table
Here's an example I found on that page too
```sql
-- A source table stored in a filesystem
CREATE TABLE Orders_in_file (
user BIGINT,
product STRING,
order_time_string STRING,
order_time AS to_timestamp(order_time)
)
PARTITIONED BY user
WITH (
'connector' = 'filesystem'
'path' = '...'
);
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478793652
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
I'll add a test for unsetting PROPERTIES.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r482527434
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
Review comment:
I'd prefer not to duplicate it in table properties. Then we would have to worry about keeping the two in sync.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479712374
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -88,4 +105,172 @@ public void testRenameTable() {
Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
tEnv.from("tl2").getSchema().getTableColumns());
}
+
+ @Test
+ public void testCreateTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ }
+
+ @Test
+ public void testCreateTableLocation() {
+ Assume.assumeFalse("HadoopCatalog does not support creating table with location", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('location'='/tmp/location')");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals("/tmp/location", table.location());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+ }
+
+ @Test
+ public void testCreatePartitionTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT, dt STRING) PARTITIONED BY(dt)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(PartitionSpec.builderFor(table.schema()).identity("dt").build(), table.spec());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).field("dt", DataTypes.STRING()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.singletonList("dt"), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testLoadTransformPartitionTable() throws TableNotExistException {
+ Schema schema = new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
+ validationCatalog.createTable(
+ TableIdentifier.of(icebergNamespace, "tl"), schema,
+ PartitionSpec.builderFor(schema).bucket("id", 100).build());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.emptyList(), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testAlterTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ tEnv.executeSql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ tEnv.executeSql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ CatalogTable catalogTable = catalogTable("tl");
+ properties.remove("oldK");
+ tEnv.getCatalog(tEnv.getCurrentCatalog()).get().alterTable(
+ new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testRelocateTable() {
+ Assume.assumeFalse("HadoopCatalog does not support relocate table", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+ tEnv.executeSql("ALTER TABLE tl SET('location'='/tmp/location')");
Review comment:
According to this documentation, the `LOCATION` clause is supported in a create table statement.
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/create.html#create-table
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479864592
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +287,29 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = getIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table getIcebergTable(ObjectPath tablePath) throws TableNotExistException {
Review comment:
OK, it is an iceberg table, change it to load.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479566449
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
Review comment:
Is there a case where `CatalogBaseTable` doesn't implement `CatalogTable`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479869298
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -88,4 +105,172 @@ public void testRenameTable() {
Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
tEnv.from("tl2").getSchema().getTableColumns());
}
+
+ @Test
+ public void testCreateTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ }
+
+ @Test
+ public void testCreateTableLocation() {
+ Assume.assumeFalse("HadoopCatalog does not support creating table with location", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('location'='/tmp/location')");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals("/tmp/location", table.location());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+ }
+
+ @Test
+ public void testCreatePartitionTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT, dt STRING) PARTITIONED BY(dt)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(PartitionSpec.builderFor(table.schema()).identity("dt").build(), table.spec());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).field("dt", DataTypes.STRING()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.singletonList("dt"), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testLoadTransformPartitionTable() throws TableNotExistException {
+ Schema schema = new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
+ validationCatalog.createTable(
+ TableIdentifier.of(icebergNamespace, "tl"), schema,
+ PartitionSpec.builderFor(schema).bucket("id", 100).build());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.emptyList(), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testAlterTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ tEnv.executeSql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ tEnv.executeSql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ CatalogTable catalogTable = catalogTable("tl");
+ properties.remove("oldK");
+ tEnv.getCatalog(tEnv.getCurrentCatalog()).get().alterTable(
+ new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testRelocateTable() {
+ Assume.assumeFalse("HadoopCatalog does not support relocate table", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+ tEnv.executeSql("ALTER TABLE tl SET('location'='/tmp/location')");
Review comment:
Yes, just the table property...
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/filesystem.html
Flink Filesystem connector also support location using `path` property.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479860810
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
Review comment:
Good point.
If there is only one operation in a single call, it seems feasible. And actually, in the SQL DDLs, only a single type is in a single SQL.
But yes, this API is unclear, if we look at it from the API level alone, there are too many possibilities...
I'll add comments in the code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478268784
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
Review comment:
nit: better to use `ImmutableMap` ?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
Review comment:
What's the reason that we could not support adding /removing/renaming column ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479985290
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +287,30 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = loadIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table loadIcebergTable(ObjectPath tablePath) throws TableNotExistException {
+ try {
+ return icebergCatalog.loadTable(toIdentifier(tablePath));
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
throw new TableNotExistException(getName(), tablePath, e);
}
}
+ private CatalogTable toCatalogTable(Table table) {
+ TableSchema schema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
Review comment:
nit: how about making the iceberg schema -> flink `TableSchema` conversion to be a static method inside FlinkSchemaUtil ? The table sink pr https://github.com/apache/iceberg/pull/1348 will also depend on this static method (https://github.com/apache/iceberg/pull/1348/files#diff-0ad7dfff9cfa32fbb760796d976fd650R50).
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +340,169 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = loadIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by comparing
+ // CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
Review comment:
Do we need to add a TODO indicating that we flink only support `identity` partition now but will support hidden column future ?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +340,169 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = loadIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by comparing
+ // CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
+ }
+ }
+ return partitionKeys;
+ }
+
+ private static void commitChanges(Table table, String setLocation, String setSnapshotId,
+ String pickSnapshotId, Map<String, String> setProperties) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is ambiguous and choosing
+ // one order leads to different results
+ Preconditions.checkArgument(setSnapshotId == null || pickSnapshotId == null,
+ "Cannot set the current snapshot ID and cherry-pick snapshot changes");
+
+ if (setSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(setSnapshotId);
+ table.manageSnapshots().setCurrentSnapshot(newSnapshotId).commit();
+ }
+
+ // if updating the table snapshot, perform that update first in case it fails
+ if (pickSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(pickSnapshotId);
+ table.manageSnapshots().cherrypick(newSnapshotId).commit();
+ }
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation()
+ .setLocation(setLocation)
+ .commit();
+ }
+
+ if (!setProperties.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ setProperties.forEach((k, v) -> {
+ if (v == null) {
Review comment:
The `v` should never be null in `HashMap` ?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +286,29 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = getIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table getIcebergTable(ObjectPath tablePath) throws TableNotExistException {
+ try {
+ return icebergCatalog.loadTable(toIdentifier(tablePath));
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
throw new TableNotExistException(getName(), tablePath, e);
}
}
+ private CatalogTable toCatalogTable(Table table) {
+ TableSchema schema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema());
+
+ // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
Review comment:
Got it, maybe could write this comment more clear.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
`setProperties.put(k, null)` ? The javadoc from Map said :
```java
* @throws NullPointerException if the specified key or value is null
* and this map does not permit null keys or values
* @throws IllegalArgumentException if some property of the specified key
* or value prevents it from being stored in this map
*/
V put(K key, V value);
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r482669447
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
Review comment:
I see, maybe we can expose these information in properties.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478288520
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
You can take a look to tests.
The existing key-values will be keep. The `newTable.getOptions()` is not just from alter DDL. It is already merged with old options.
Actually, there is not DDL to delete key-value too...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479865130
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
Review comment:
It is `CatalogView`, Iceberg catalog does not support views, so if there is a view, should be a bug...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r480004115
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +287,30 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = loadIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table loadIcebergTable(ObjectPath tablePath) throws TableNotExistException {
+ try {
+ return icebergCatalog.loadTable(toIdentifier(tablePath));
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
throw new TableNotExistException(getName(), tablePath, e);
}
}
+ private CatalogTable toCatalogTable(Table table) {
+ TableSchema schema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
Review comment:
static looks good to me, but I think we can keep it in this class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479712692
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -88,4 +105,172 @@ public void testRenameTable() {
Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
tEnv.from("tl2").getSchema().getTableColumns());
}
+
+ @Test
+ public void testCreateTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ }
+
+ @Test
+ public void testCreateTableLocation() {
+ Assume.assumeFalse("HadoopCatalog does not support creating table with location", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('location'='/tmp/location')");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals("/tmp/location", table.location());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+ }
+
+ @Test
+ public void testCreatePartitionTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT, dt STRING) PARTITIONED BY(dt)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(PartitionSpec.builderFor(table.schema()).identity("dt").build(), table.spec());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).field("dt", DataTypes.STRING()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.singletonList("dt"), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testLoadTransformPartitionTable() throws TableNotExistException {
+ Schema schema = new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
+ validationCatalog.createTable(
+ TableIdentifier.of(icebergNamespace, "tl"), schema,
+ PartitionSpec.builderFor(schema).bucket("id", 100).build());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.emptyList(), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testAlterTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ tEnv.executeSql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ tEnv.executeSql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ CatalogTable catalogTable = catalogTable("tl");
+ properties.remove("oldK");
+ tEnv.getCatalog(tEnv.getCurrentCatalog()).get().alterTable(
+ new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testRelocateTable() {
+ Assume.assumeFalse("HadoopCatalog does not support relocate table", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+ tEnv.executeSql("ALTER TABLE tl SET('location'='/tmp/location')");
Review comment:
However, looking forward at the Flink 1.12 prerelease docs, it appears that there's added support for a `Hive Dialect` which supports `LOCATION` clause.
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/hive/hive_dialect.html#create-1
```hiveql
CREATE [EXTERNAL] TABLE [IF NOT EXISTS] table_name
[(col_name data_type [column_constraint] [COMMENT col_comment], ... [table_constraint])]
[COMMENT table_comment]
[PARTITIONED BY (col_name data_type [COMMENT col_comment], ...)]
[
[ROW FORMAT row_format]
[STORED AS file_format]
]
[LOCATION fs_path]
[TBLPROPERTIES (property_name=property_value, ...)]
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479710154
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +287,29 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = getIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table getIcebergTable(ObjectPath tablePath) throws TableNotExistException {
Review comment:
I agree on the `get`, but it does align with the interface's `getTable` method which this is used in. However, it also calls out to `load` so the argument that the iceberg specific stuff might stick with `load` vs `get` is still valid.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479712374
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -88,4 +105,172 @@ public void testRenameTable() {
Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
tEnv.from("tl2").getSchema().getTableColumns());
}
+
+ @Test
+ public void testCreateTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ }
+
+ @Test
+ public void testCreateTableLocation() {
+ Assume.assumeFalse("HadoopCatalog does not support creating table with location", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('location'='/tmp/location')");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals("/tmp/location", table.location());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+ }
+
+ @Test
+ public void testCreatePartitionTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT, dt STRING) PARTITIONED BY(dt)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(PartitionSpec.builderFor(table.schema()).identity("dt").build(), table.spec());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).field("dt", DataTypes.STRING()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.singletonList("dt"), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testLoadTransformPartitionTable() throws TableNotExistException {
+ Schema schema = new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
+ validationCatalog.createTable(
+ TableIdentifier.of(icebergNamespace, "tl"), schema,
+ PartitionSpec.builderFor(schema).bucket("id", 100).build());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.emptyList(), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testAlterTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ tEnv.executeSql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ tEnv.executeSql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ CatalogTable catalogTable = catalogTable("tl");
+ properties.remove("oldK");
+ tEnv.getCatalog(tEnv.getCurrentCatalog()).get().alterTable(
+ new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testRelocateTable() {
+ Assume.assumeFalse("HadoopCatalog does not support relocate table", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+ tEnv.executeSql("ALTER TABLE tl SET('location'='/tmp/location')");
Review comment:
EDIT: In the CREATE TABLE documentation, I cannot find a reference to `LOCATION` outside of table properties: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/create.html#create-table
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478284990
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
Q: Does this align with the flink sql semantics ?
I saw the [document](https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/alter.html#alter-table) said: "Set one or more properties in the specified table. If a particular property is already set in the table, override the old value with the new one."
```sql
ALTER TABLE [catalog_name.][db_name.]table_name SET (key1=val1, key2=val2, ...)
```
For the existing key-values (in old table ) which don't appear in the new table, should we remove them from old table ? ( The document did not describe this case clearly, just for confirmation).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479862586
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
But properties updates does not have column IDs. As long as the last is the same.
Sorry, I don't get your point.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479864592
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +287,29 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = getIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table getIcebergTable(ObjectPath tablePath) throws TableNotExistException {
Review comment:
It is OK to change it to load.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r482529029
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
Review comment:
I was thinking that since operations are delegated to Iceberg, correctness is not an issue. It would be nice to show users which columns are partition columns so they can see which ones are good candidates for query predicates. I don't think this is a blocker, though.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi edited a comment on pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi edited a comment on pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#issuecomment-683516342
Thanks a lot for your review~ @rdblue and @kbendick I'll take a look~
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r482526209
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
+ }
+ }
+ return partitionKeys;
+ }
+
+ private static void commitChanges(Table table, String setLocation, String setSnapshotId,
+ String pickSnapshotId, Map<String, String> setProperties) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is ambiguous and choosing
+ // one order leads to different results
+ Preconditions.checkArgument(setSnapshotId == null || pickSnapshotId == null,
+ "Cannot set the current the current snapshot ID and cherry-pick snapshot changes");
+
+ if (setSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(setSnapshotId);
+ table.manageSnapshots().setCurrentSnapshot(newSnapshotId).commit();
+ }
+
+ // if updating the table snapshot, perform that update first in case it fails
+ if (pickSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(pickSnapshotId);
+ table.manageSnapshots().cherrypick(newSnapshotId).commit();
+ }
+
+ Transaction transaction = table.newTransaction();
Review comment:
That explains it. Thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479710309
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
Review comment:
Should `location` still be placed in the table properties or will that cause some kind of conflict / error?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
Review comment:
To me, it seems like adding all of the identity fields (but not the transformed fields) would likely be incorrect. Although returning an empty list when the table is partitioned seems like a possible correctness bug to me too.
Should we consider throwing an exception in this case instead until such a time that Flink supports partition transforms?
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
Review comment:
`CatalogTable` is a subinterface that inherits from `CatalogBaseTable`. So definitely, yes.
See the java docs on the current `CatalogBaseTable` in Flink:
https://ci.apache.org/projects/flink/flink-docs-release-1.11/api/java/org/apache/flink/table/catalog/CatalogBaseTable.html
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
+ }
+ }
+ return partitionKeys;
+ }
+
+ private static void commitChanges(Table table, String setLocation, String setSnapshotId,
+ String pickSnapshotId, Map<String, String> setProperties) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is ambiguous and choosing
+ // one order leads to different results
+ Preconditions.checkArgument(setSnapshotId == null || pickSnapshotId == null,
+ "Cannot set the current the current snapshot ID and cherry-pick snapshot changes");
Review comment:
Nit: Duplication of the words `the current` in the Preconditions string. It currently reads `Cannot set the current the current`.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
Review comment:
I found this which might answer your question: https://cwiki.apache.org/confluence/display/FLINK/FLIP+87%3A+Primary+key+constraints+in+Table+API
In particular, here are the _proposed_ changes:
```
Proposed Changes
We suggest to introduce the concept of primary key constraint as a hint for FLINK to leverage for optimizations.
Primary key constraints tell that a column or a set of columns of a table or a view are unique and they do not contain null.
Neither of columns in a primary can be nullable.
Primary key therefore uniquely identify a row in a table.
```
So it sounds just like an RDBMS primary key.
Note however, that even in the FLIP (which is just the proposal and not necessarily the finished product), it does state that there's no planned enforcement on the PK. It's up to the user to ensure that the PK is non-null and unique.
```
Primary key validity checks
SQL standard specifies that a constraint can either be ENFORCED or NOT ENFORCED.
This controls if the constraint checks are performed on the incoming/outgoing data.
Flink does not own the data therefore the only mode we want to support is the NOT ENFORCED mode.
Its up to the user to ensure that the query enforces key integrity.
```
So I agree here that throwing might be the most useful option and that there's likely nothing on the iceberg side to be added to enforce this as Flink doesn't enforce it either. In an entirely streaming setting, ensuring unique keys would be rather difficult and so to me it somewhat sounds like the PK is just more metadata that could very well be in TBLPROPERTIES.
But a more experienced Flink SQL user than myself might have more to say on the matter. I've never attempted to enforce a PK when using Flink SQL. Sounds like the work to do so would involve custom operators etc.
TLDR: The Primary Key is just a constraint, which is currently part of Flink's Table spec but goes unenforced and is up to the user. It does not appear as though the PK info is supported in any UpsertSinks etc, though that may be discussed / planned in the future. Support in the DDL for Primary Key constraints is relatively new (Flink 1.11 / current, with support in the API coming in at Flink 1.10).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479566628
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
Review comment:
Is this something we should add to Iceberg for Flink use cases? What does Flink use the primary key for?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r480005436
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +340,169 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = loadIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by comparing
+ // CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
Review comment:
Already have todo in `toPartitionKeys`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r480006570
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +340,169 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = loadIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by comparing
+ // CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
+ }
+ }
+ return partitionKeys;
+ }
+
+ private static void commitChanges(Table table, String setLocation, String setSnapshotId,
+ String pickSnapshotId, Map<String, String> setProperties) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is ambiguous and choosing
+ // one order leads to different results
+ Preconditions.checkArgument(setSnapshotId == null || pickSnapshotId == null,
+ "Cannot set the current snapshot ID and cherry-pick snapshot changes");
+
+ if (setSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(setSnapshotId);
+ table.manageSnapshots().setCurrentSnapshot(newSnapshotId).commit();
+ }
+
+ // if updating the table snapshot, perform that update first in case it fails
+ if (pickSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(pickSnapshotId);
+ table.manageSnapshots().cherrypick(newSnapshotId).commit();
+ }
+
+ Transaction transaction = table.newTransaction();
+
+ if (setLocation != null) {
+ transaction.updateLocation()
+ .setLocation(setLocation)
+ .commit();
+ }
+
+ if (!setProperties.isEmpty()) {
+ UpdateProperties updateProperties = transaction.updateProperties();
+ setProperties.forEach((k, v) -> {
+ if (v == null) {
Review comment:
`HashMap` allows nulls:
```
* Hash table based implementation of the <tt>Map</tt> interface. This
* implementation provides all of the optional map operations, and permits
* <tt>null</tt> values and the <tt>null</tt> key. (The <tt>HashMap</tt>
* class is roughly equivalent to <tt>Hashtable</tt>, except that it is
* unsynchronized and permits nulls.)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r482524417
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
My point is that there isn't a correctness problem so this is okay. But, this causes Flink to do much more work because it has to apply changes from SQL, then recover those changes by comparing property maps, and pass the changes to Iceberg so that Iceberg can apply the changes. It is easier to pass the changes directly to Iceberg if the Flink API can be updated to support it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478296181
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
> It is already merged with old options.
If the new options are already merged with old options, so for the key in old options, shouldn't it be always in new options ? Seems there's no reason to add this sentence here ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r482667962
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
Got it, I think we can have a try in Flink.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479866703
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
Review comment:
Thanks @kbendick .
At present, in Flink, PK is mainly used to process CDC stream.
-For example, if user access a Kafka source of a CDC stream, the user can define a primary key. In this way, the downstream can perform efficient dynamic table / static table conversion (restore to the original static table) according to a certain primary key.
-For example, when the stream data (CDC) is written into a JDBC database, the user can define primary key. In this way, Flink can insert the data into the database by using the upsert writing way.
I think, If iceberg supports CDC native processing in the future, we may be able to use it.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
Review comment:
Thanks @kbendick .
At present, in Flink, PK is mainly used to process CDC stream.
- For example, if user access a Kafka source of a CDC stream, the user can define a primary key. In this way, the downstream can perform efficient dynamic table / static table conversion (restore to the original static table) according to a certain primary key.
- For example, when the stream data (CDC) is written into a JDBC database, the user can define primary key. In this way, Flink can insert the data into the database by using the upsert writing way.
I think, If iceberg supports CDC native processing in the future, we may be able to use it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479563986
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +287,29 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = getIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table getIcebergTable(ObjectPath tablePath) throws TableNotExistException {
Review comment:
`get` is not a very specific verb. I usually prefer `load` for cases like this because it more accurately describes what is happening.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479869458
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -88,4 +105,172 @@ public void testRenameTable() {
Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
tEnv.from("tl2").getSchema().getTableColumns());
}
+
+ @Test
+ public void testCreateTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ }
+
+ @Test
+ public void testCreateTableLocation() {
+ Assume.assumeFalse("HadoopCatalog does not support creating table with location", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('location'='/tmp/location')");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals("/tmp/location", table.location());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+ }
+
+ @Test
+ public void testCreatePartitionTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT, dt STRING) PARTITIONED BY(dt)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(PartitionSpec.builderFor(table.schema()).identity("dt").build(), table.spec());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).field("dt", DataTypes.STRING()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.singletonList("dt"), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testLoadTransformPartitionTable() throws TableNotExistException {
+ Schema schema = new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
+ validationCatalog.createTable(
+ TableIdentifier.of(icebergNamespace, "tl"), schema,
+ PartitionSpec.builderFor(schema).bucket("id", 100).build());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.emptyList(), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testAlterTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ tEnv.executeSql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ tEnv.executeSql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ CatalogTable catalogTable = catalogTable("tl");
+ properties.remove("oldK");
+ tEnv.getCatalog(tEnv.getCurrentCatalog()).get().alterTable(
+ new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testRelocateTable() {
+ Assume.assumeFalse("HadoopCatalog does not support relocate table", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+ tEnv.executeSql("ALTER TABLE tl SET('location'='/tmp/location')");
Review comment:
The Hive dialect should only works for Hive tables. There are lots of tricky things..
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479867574
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
Review comment:
+1 to likely be incorrect.
But I don't want to throw an exception because it can read the existing iceberg table.
Actually, the returned partition keys are useless, except that Flink can show users the meta information of the table.
All partition operations are directly delegated to specific source / sink, so Flink does not need to see partition information.
I tend to support it so that we can read existing Iceberg tables.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479567020
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
Review comment:
All tables with any partition transform other than identity appear to be unpartitioned? Why not return all of the identity fields at least?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#issuecomment-683181736
This looks mostly good to me. I just have a few questions.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479862586
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
But properties updates does not have column IDs. As long as the last is the same.
Sorry, I didn't get your point.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479567639
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
##########
@@ -88,4 +105,172 @@ public void testRenameTable() {
Collections.singletonList(TableColumn.of("id", DataTypes.BIGINT())),
tEnv.from("tl2").getSchema().getTableColumns());
}
+
+ @Test
+ public void testCreateTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ }
+
+ @Test
+ public void testCreateTableLocation() {
+ Assume.assumeFalse("HadoopCatalog does not support creating table with location", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('location'='/tmp/location')");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(Types.NestedField.optional(1, "id", Types.LongType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals("/tmp/location", table.location());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+ }
+
+ @Test
+ public void testCreatePartitionTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT, dt STRING) PARTITIONED BY(dt)");
+
+ Table table = table("tl");
+ Assert.assertEquals(
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "dt", Types.StringType.get())).asStruct(),
+ table.schema().asStruct());
+ Assert.assertEquals(PartitionSpec.builderFor(table.schema()).identity("dt").build(), table.spec());
+ Assert.assertEquals(Maps.newHashMap(), table.properties());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).field("dt", DataTypes.STRING()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.singletonList("dt"), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testLoadTransformPartitionTable() throws TableNotExistException {
+ Schema schema = new Schema(Types.NestedField.optional(0, "id", Types.LongType.get()));
+ validationCatalog.createTable(
+ TableIdentifier.of(icebergNamespace, "tl"), schema,
+ PartitionSpec.builderFor(schema).bucket("id", 100).build());
+
+ CatalogTable catalogTable = catalogTable("tl");
+ Assert.assertEquals(
+ TableSchema.builder().field("id", DataTypes.BIGINT()).build(),
+ catalogTable.getSchema());
+ Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
+ Assert.assertEquals(Collections.emptyList(), catalogTable.getPartitionKeys());
+ }
+
+ @Test
+ public void testAlterTable() throws TableNotExistException {
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT) WITH ('oldK'='oldV')");
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("oldK", "oldV");
+
+ // new
+ tEnv.executeSql("ALTER TABLE tl SET('newK'='newV')");
+ properties.put("newK", "newV");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // update old
+ tEnv.executeSql("ALTER TABLE tl SET('oldK'='oldV2')");
+ properties.put("oldK", "oldV2");
+ Assert.assertEquals(properties, table("tl").properties());
+
+ // remove property
+ CatalogTable catalogTable = catalogTable("tl");
+ properties.remove("oldK");
+ tEnv.getCatalog(tEnv.getCurrentCatalog()).get().alterTable(
+ new ObjectPath(DATABASE, "tl"), catalogTable.copy(properties), false);
+ Assert.assertEquals(properties, table("tl").properties());
+ }
+
+ @Test
+ public void testRelocateTable() {
+ Assume.assumeFalse("HadoopCatalog does not support relocate table", isHadoopCatalog);
+
+ tEnv.executeSql("CREATE TABLE tl(id BIGINT)");
+ tEnv.executeSql("ALTER TABLE tl SET('location'='/tmp/location')");
Review comment:
Does Flink support the `LOCATION` clause, or just the table property?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478323128
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
What do you mean?
The new options is new all options.
For example:
old: {'j' = 'am', 'p' = 'an'}
alter: `ALTER TABLE t UNSET TBLPROPERTIES ('j')`
`newTable.getOptions()` will be: {'p' = 'an'}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479868725
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +339,167 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+
+ ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
+ String location = null;
+ for (Map.Entry<String, String> entry : table.getOptions().entrySet()) {
+ if ("location".equalsIgnoreCase(entry.getKey())) {
+ location = entry.getValue();
+ } else {
+ properties.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ location,
+ properties.build());
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
+ }
+ });
+
+ commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
+ }
+
+ private static void validateFlinkTable(CatalogBaseTable table) {
+ Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
+
+ TableSchema schema = table.getSchema();
+ schema.getTableColumns().forEach(column -> {
+ if (column.isGenerated()) {
+ throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
+ }
+ });
+
+ if (!schema.getWatermarkSpecs().isEmpty()) {
+ throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
+ }
+
+ if (schema.getPrimaryKey().isPresent()) {
+ throw new UnsupportedOperationException("Creating table with primary key is not supported yet.");
+ }
+ }
+
+ private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
+ PartitionSpec.Builder builder = PartitionSpec.builderFor(icebergSchema);
+ partitionKeys.forEach(builder::identity);
+ return builder.build();
+ }
+
+ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSchema) {
+ List<String> partitionKeys = Lists.newArrayList();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().isIdentity()) {
+ partitionKeys.add(icebergSchema.findColumnName(field.sourceId()));
+ } else {
+ // Not created by Flink SQL.
+ // For compatibility with iceberg tables, return empty.
+ // TODO modify this after Flink support partition transform.
+ return Collections.emptyList();
+ }
+ }
+ return partitionKeys;
+ }
+
+ private static void commitChanges(Table table, String setLocation, String setSnapshotId,
+ String pickSnapshotId, Map<String, String> setProperties) {
+ // don't allow setting the snapshot and picking a commit at the same time because order is ambiguous and choosing
+ // one order leads to different results
+ Preconditions.checkArgument(setSnapshotId == null || pickSnapshotId == null,
+ "Cannot set the current the current snapshot ID and cherry-pick snapshot changes");
+
+ if (setSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(setSnapshotId);
+ table.manageSnapshots().setCurrentSnapshot(newSnapshotId).commit();
+ }
+
+ // if updating the table snapshot, perform that update first in case it fails
+ if (pickSnapshotId != null) {
+ long newSnapshotId = Long.parseLong(pickSnapshotId);
+ table.manageSnapshots().cherrypick(newSnapshotId).commit();
+ }
+
+ Transaction transaction = table.newTransaction();
Review comment:
Looks like the `manageSnapshots` is unsupported in `TransactionTable`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] openinx commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478301144
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -277,20 +286,29 @@ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean igno
}
@Override
- public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
- try {
- Table table = icebergCatalog.loadTable(toIdentifier(tablePath));
- TableSchema tableSchema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ public CatalogTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
+ Table table = getIcebergTable(tablePath);
+ return toCatalogTable(table);
+ }
- // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
- // catalog table.
- // Let's re-loading table from Iceberg catalog when creating source/sink operators.
- return new CatalogTableImpl(tableSchema, table.properties(), null);
+ private Table getIcebergTable(ObjectPath tablePath) throws TableNotExistException {
+ try {
+ return icebergCatalog.loadTable(toIdentifier(tablePath));
} catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
throw new TableNotExistException(getName(), tablePath, e);
}
}
+ private CatalogTable toCatalogTable(Table table) {
+ TableSchema schema = FlinkSchemaUtil.toSchema(FlinkSchemaUtil.convert(table.schema()));
+ List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema());
+
+ // NOTE: We can not create a IcebergCatalogTable, because Flink optimizer may use CatalogTableImpl to copy a new
Review comment:
nit: we may need to change this comment ? Actually, I did not found any class named `IcebergCatalogTable`, or I misunderstood something ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] JingsongLi commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r478273911
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
Review comment:
No Flink DLL to add/removing/renaming column...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1393: Flink: Support creating table and altering table in Flink SQL
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1393:
URL: https://github.com/apache/iceberg/pull/1393#discussion_r479566299
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
##########
@@ -320,19 +338,158 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor
}
}
- /**
- * TODO Add partitioning to the Flink DDL parser.
- */
@Override
public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support createTable now.");
+ throws CatalogException, TableAlreadyExistException {
+ validateFlinkTable(table);
+
+ Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+ PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
+ Map<String, String> options = Maps.newHashMap(table.getOptions());
+
+ try {
+ icebergCatalog.createTable(
+ toIdentifier(tablePath),
+ icebergSchema,
+ spec,
+ options.get("location"),
+ options);
+ } catch (AlreadyExistsException e) {
+ throw new TableAlreadyExistException(getName(), tablePath, e);
+ }
}
@Override
public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
- throws CatalogException {
- throw new UnsupportedOperationException("Not support alterTable now.");
+ throws CatalogException, TableNotExistException {
+ validateFlinkTable(newTable);
+ Table icebergTable = getIcebergTable(tablePath);
+ CatalogTable table = toCatalogTable(icebergTable);
+
+ // Currently, Flink SQL only support altering table properties.
+
+ if (!table.getSchema().equals(newTable.getSchema())) {
+ throw new UnsupportedOperationException("Altering schema is not supported yet.");
+ }
+
+ if (!table.getPartitionKeys().equals(((CatalogTable) newTable).getPartitionKeys())) {
+ throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+ }
+
+ Map<String, String> oldOptions = table.getOptions();
+ Map<String, String> setProperties = Maps.newHashMap();
+
+ String setLocation = null;
+ String setSnapshotId = null;
+ String pickSnapshotId = null;
+
+ for (Map.Entry<String, String> entry : newTable.getOptions().entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+
+ if (Objects.equals(value, oldOptions.get(key))) {
+ continue;
+ }
+
+ if ("location".equalsIgnoreCase(key)) {
+ setLocation = value;
+ } else if ("current-snapshot-id".equalsIgnoreCase(key)) {
+ setSnapshotId = value;
+ } else if ("cherry-pick-snapshot-id".equalsIgnoreCase(key)) {
+ pickSnapshotId = value;
+ } else {
+ setProperties.put(key, value);
+ }
+ }
+
+ oldOptions.keySet().forEach(k -> {
+ if (!newTable.getOptions().containsKey(k)) {
+ setProperties.put(k, null);
Review comment:
I think it is okay to diff the property sets like this, but it seems like it would be easier not to. Right now, Flink has to apply the changes, then this code diffs the property sets, then Iceberg will re-apply the changes. In addition, this model doesn't work for schema updates, as I noted above.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org