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 2021/06/03 03:19:58 UTC
[GitHub] [iceberg] openinx opened a new pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
openinx opened a new pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666
This PR is trying to address the issue : https://github.com/apache/iceberg/issues/2572
--
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] gaborgsomogyi commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r650113649
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
Review comment:
Nit: Extra newline can be removed.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
+ .build();
+
+ tEnv = TableEnvironment.create(settings);
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+
+ @Test
+ public void testHadoopCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "local_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ sql("INSERT INTO hadoop_table VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')");
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM hadoop_table"));
+
+ // Drop and create it again.
+ sql("DROP TABLE hadoop_table");
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM hadoop_table"));
+
+ sql("DROP TABLE hadoop_table");
+ }
+
+ @Test
+ public void testHiveCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hive");
+ tableProps.put("catalog-type", "hive");
+ tableProps.put("catalog-database", "default");
+ tableProps.put(CatalogProperties.URI, FlinkCatalogTestBase.getURI(hiveConf));
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hive_table(id BIGINT, data STRING) WITH %s", FlinkCatalogTestBase.toWithClause(tableProps));
+ sql("INSERT INTO hive_table VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')");
+ Assert.assertEquals("Should have the expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM hive_table"));
+
+ // Drop and create it again.
+ sql("DROP TABLE hive_table");
+ sql("CREATE TABLE hive_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM hive_table"));
+
+ sql("DROP TABLE hive_table");
+ }
+
+ private String toWithClause(Map<String, String> props) {
+ return FlinkCatalogTestBase.toWithClause(props);
+ }
+
+ private String warehouseRoot() {
+ return String.format("file://%s", warehouse.getRoot().getAbsolutePath());
+ }
+
+ protected List<Row> sql(String query, Object... args) {
Review comment:
+1 on double checking whether it's needed at all. When I've removed the test still passed.
If there is an objective reason why this needed the `@Override` must be added.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
Review comment:
I've double checked it works with `inStreamingMode()`. It would be good to add parameterized tests which runs in both modes.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,150 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ public static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ public static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type.");
+
+ public static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_VALUE)
+ .withDeprecatedKeys("Catalog database");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
Review comment:
I've double checked and when default constructor removed the tests fail with the mentioned issue so it's needed.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,143 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_VALUE)
+ .withDeprecatedKeys("Catalog database");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ options.add(CATALOG_DATABASE);
+ return options;
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return Sets.newHashSet();
}
@Override
public String factoryIdentifier() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return FACTORY_IDENTIFIER;
+ }
+
+ private static TableLoader createTableLoader(CatalogBaseTable catalogTable,
+ Map<String, String> tableProps,
+ String tableName) {
+ Configuration flinkConf = new Configuration();
+ tableProps.forEach(flinkConf::setString);
+
+ String catalogName = flinkConf.getString(CATALOG_NAME);
+ Preconditions.checkNotNull(catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());
+
+ String catalogDatabase = flinkConf.getString(CATALOG_DATABASE);
+ Preconditions.checkNotNull(catalogDatabase, "Table property '%s' cannot be null", CATALOG_DATABASE.key());
+
+ org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
+ CatalogLoader catalogLoader = FlinkCatalogFactory.createCatalogLoader(catalogName, tableProps, hadoopConf);
Review comment:
@stevenzwu please see up. In this PR `createCatalogLoader` is made static. +1 not making static.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
+ .build();
+
+ tEnv = TableEnvironment.create(settings);
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+
+ @Test
+ public void testHadoopCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "local_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ sql("INSERT INTO hadoop_table VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')");
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM hadoop_table"));
+
+ // Drop and create it again.
+ sql("DROP TABLE hadoop_table");
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM hadoop_table"));
+
+ sql("DROP TABLE hadoop_table");
+ }
+
+ @Test
+ public void testHiveCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hive");
+ tableProps.put("catalog-type", "hive");
+ tableProps.put("catalog-database", "default");
+ tableProps.put(CatalogProperties.URI, FlinkCatalogTestBase.getURI(hiveConf));
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hive_table(id BIGINT, data STRING) WITH %s", FlinkCatalogTestBase.toWithClause(tableProps));
+ sql("INSERT INTO hive_table VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')");
+ Assert.assertEquals("Should have the expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM hive_table"));
+
+ // Drop and create it again.
+ sql("DROP TABLE hive_table");
+ sql("CREATE TABLE hive_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM hive_table"));
+
+ sql("DROP TABLE hive_table");
+ }
+
+ private String toWithClause(Map<String, String> props) {
+ return FlinkCatalogTestBase.toWithClause(props);
+ }
+
+ private String warehouseRoot() {
+ return String.format("file://%s", warehouse.getRoot().getAbsolutePath());
+ }
+
+ protected List<Row> sql(String query, Object... args) {
Review comment:
+1 on double checking whether it's needed at all. When I've removed the test still passed.
If there is an objective reason why this needed then `@Override` must be added.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
+ .build();
+
+ tEnv = TableEnvironment.create(settings);
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+
+ @Test
+ public void testHadoopCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "local_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ sql("INSERT INTO hadoop_table VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')");
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM hadoop_table"));
+
+ // Drop and create it again.
+ sql("DROP TABLE hadoop_table");
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM hadoop_table"));
+
+ sql("DROP TABLE hadoop_table");
+ }
+
+ @Test
+ public void testHiveCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hive");
+ tableProps.put("catalog-type", "hive");
+ tableProps.put("catalog-database", "default");
+ tableProps.put(CatalogProperties.URI, FlinkCatalogTestBase.getURI(hiveConf));
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hive_table(id BIGINT, data STRING) WITH %s", FlinkCatalogTestBase.toWithClause(tableProps));
+ sql("INSERT INTO hive_table VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')");
+ Assert.assertEquals("Should have the expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM hive_table"));
+
+ // Drop and create it again.
+ sql("DROP TABLE hive_table");
+ sql("CREATE TABLE hive_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM hive_table"));
+
+ sql("DROP TABLE hive_table");
+ }
+
+ private String toWithClause(Map<String, String> props) {
+ return FlinkCatalogTestBase.toWithClause(props);
+ }
+
+ private String warehouseRoot() {
+ return String.format("file://%s", warehouse.getRoot().getAbsolutePath());
+ }
+
+ protected List<Row> sql(String query, Object... args) {
Review comment:
+1 on double checking whether it's needed at all. When I've removed the whole function then the test still passed.
If there is an objective reason why this needed then `@Override` must be added.
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r645737621
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,150 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ public static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ public static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type.");
+
+ public static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_VALUE)
+ .withDeprecatedKeys("Catalog database");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ options.add(CATALOG_DATABASE);
+ return Sets.newHashSet();
Review comment:
Are you intending to return `options` here? Possibly I am not understanding the interface fully, but it seems like `options` should be returned.
--
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] stevenzwu commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r647117231
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
+ .build();
+
+ tEnv = TableEnvironment.create(settings);
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+
+ @Test
+ public void testHadoopCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "local_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ sql("INSERT INTO hadoop_table VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')");
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM hadoop_table"));
+
+ // Drop and create it again.
+ sql("DROP TABLE hadoop_table");
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM hadoop_table"));
+
+ sql("DROP TABLE hadoop_table");
+ }
+
+ @Test
+ public void testHiveCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hive");
+ tableProps.put("catalog-type", "hive");
+ tableProps.put("catalog-database", "default");
+ tableProps.put(CatalogProperties.URI, FlinkCatalogTestBase.getURI(hiveConf));
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hive_table(id BIGINT, data STRING) WITH %s", FlinkCatalogTestBase.toWithClause(tableProps));
+ sql("INSERT INTO hive_table VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')");
+ Assert.assertEquals("Should have the expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM hive_table"));
+
+ // Drop and create it again.
+ sql("DROP TABLE hive_table");
+ sql("CREATE TABLE hive_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM hive_table"));
+
+ sql("DROP TABLE hive_table");
+ }
+
+ private String toWithClause(Map<String, String> props) {
+ return FlinkCatalogTestBase.toWithClause(props);
+ }
+
+ private String warehouseRoot() {
+ return String.format("file://%s", warehouse.getRoot().getAbsolutePath());
+ }
+
+ protected List<Row> sql(String query, Object... args) {
Review comment:
`FlinkTestBase` already has a `sql` method. it seems that the only diff is the extra call of `tableResult.await()` here. Is it necessary considering we are calling `tableResult.collect()`?
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-882946191
@openinx, I'll take another look at this. Is there a description of the proposed behavior for these tables in the Flink catalog?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659467383
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
I think this comment is similar to the @stevenzwu 's [question](https://github.com/apache/iceberg/pull/2666#discussion_r647120788) . Saying if the flink table identifier is `flink_catalog.flink_database.table_name` (when creating table using the `connector=iceberg`, not the iceberg catalog approach), then it's mapping the flink table with name `table_name` to the underlying iceberg table `table_name` with the configured database in table property. The flink's `flink_catalog` and `flink_database` does not has any relationship to the iceberg's catalog & database.
Maybe it's better to forbidden people to specify a `connector=iceberg` property when creating table under the iceberg catalog ( adding a Precondition.checkArgument in `createDynamicTableSource` and `createDynamicTableSink`), then we don't mixed the two approaches to create flink+iceberg table or map flink table to iceberg 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660569612
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @ClassRule
+ public static final TemporaryFolder WAREHOUSE = new TemporaryFolder();
+
+ private final String catalogName;
+ private final Map<String, String> properties;
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "catalogName = {0}, properties = {1}, isStreaming={2}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ // Create iceberg table in the hadoop catalog and default database.
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop"
+ ),
+ true
+ },
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop"
+ ),
+ false
+ },
+ // Create iceberg table in the hadoop catalog and not_existing_db.
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop",
+ "catalog-database", "not_existing_db"
+ ),
+ true
+ },
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop",
+ "catalog-database", "not_existing_db"
+ ),
+ false
+ },
+ // Create iceberg table in the hive catalog and default database.
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive"
+ ),
+ true
+ },
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive"
+ ),
+ false
+ },
+ // Create iceberg table in the hive catalog and not_existing_db.
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive",
+ "catalog-database", "not_existing_db"
+ ),
+ true
+ },
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive",
+ "catalog-database", "not_existing_db"
+ ),
+ false
+ }
+ );
+ }
+
+ public TestIcebergConnector(String catalogName, Map<String, String> properties, boolean isStreaming) {
+ this.catalogName = catalogName;
+ this.properties = properties;
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() throws TException {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+
+ // Clean the created orphan databases and tables from hive-metastore.
+ if (isHiveCatalog()) {
+ HiveMetaStoreClient metaStoreClient = new HiveMetaStoreClient(hiveConf);
+ try {
+ metaStoreClient.dropTable(databaseName(), TABLE_NAME);
+ if (!isDefaultDatabaseName()) {
+ try {
+ metaStoreClient.dropDatabase(databaseName());
+ } catch (Exception ignored) {
+ // Ignore
+ }
+ }
+ } finally {
+ metaStoreClient.close();
+ }
+ }
+ }
+
+ private void testCreateConnectorTable() {
+ Map<String, String> tableProps = createTableProps();
+
+ // Create table under the flink's current database.
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ Catalog flinkCatalog = factory.createCatalog(catalogName, tableProps, new Configuration());
+ Assert.assertTrue("Should have created the expected database", flinkCatalog.databaseExists(databaseName()));
+ Assert.assertTrue("Should have created the expected table",
+ flinkCatalog.tableExists(new ObjectPath(databaseName(), TABLE_NAME)));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
Review comment:
Unfortunately, the flink sql don't provide the syntax to support it now, although I agree it's a better user-experience . Currently, the flink sql provide the [CREATE TABLE a LIKE b](https://iceberg.apache.org/flink/#create-table-like), but it still require that the table b is a flink catalog 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-866996230
@openinx, sorry for the delay. I'll make some time to review this.
I also know that while I was out there were a few PRs that I wasn't able to help move along. If you have a list of PRs that are important for you to get in, please send them to me and I'll make time to get them reviewed as I can. Thank you for being patient with me!
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660201187
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
Okay, I thought that this was a way to run DDL even if you don't have an Iceberg catalog defined. It sort of does that, but it also creates a reference in the in-memory catalog. That's fine, but it does bring up a couple other questions:
* How do you create an in-memory catalog table pointing to an Iceberg table if the Iceberg table already exists? What if the DDL, like the schema, doesn't match?
* Why share the table name between the in-memory and external catalog but not the database? I think it makes sense to default the external database and table name using the ones from the DDL command but allow both to be overridden.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659090803
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "isStreaming={0}")
Review comment:
Could the catalog be a parameter as well? We do that in [Spark tests](https://github.com/apache/iceberg/blob/master/spark3/src/test/java/org/apache/iceberg/spark/SparkCatalogTestBase.java#L57-L76) and it works well.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660565409
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
> Why share the table name between the in-memory and external catalog but not the database?
I think you are right. I checked the [jdbc](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/jdbc/) connector, we could specify a different jdbc table name when creating the flink table. It make sense to allow both db & table name to be overridden.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] gaborgsomogyi commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r655146166
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
Review comment:
OK, this is nice to have.
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660553618
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
Review comment:
I got your question, yes, it's possible in your case. For example, when we create a table under the iceberg catalog in flink sql, then it will create the `FlinkDynamicTableFactory` with the specified iceberg catalog.
There should be no problem in that case. the `catalogTable` is the correct flink table which is parsed from SQL such as `INSERT INTO iceberg_catalog.iceberg_db.iceberg_table` , although we don't use it in the current code 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r646564200
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,150 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ public static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ public static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type.");
+
+ public static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_VALUE)
+ .withDeprecatedKeys("Catalog database");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ options.add(CATALOG_DATABASE);
+ return Sets.newHashSet();
Review comment:
You are correct, we will need to return the `options`. Thanks for the checking.
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660201187
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
Okay, I thought that this was a way to run DDL even if you don't have an Iceberg catalog defined. It sort of does that, but it also creates a reference in the in-memory catalog. That's fine, but it does bring up a couple other questions:
* How do you create an in-memory catalog table pointing to an Iceberg table if the Iceberg table already exists? What if the DDL, like the schema, doesn't match?
* Why share the table name between the in-memory and external catalog but not the database? I think it makes sense to default the external database and table name using the ones from the DDL command but allow both to be overridden.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
Review comment:
Sorry, I meant is there a case when `context.getCatalogTable()` is non-null but passed to a source where `catalog` is set in the constructor?
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @ClassRule
+ public static final TemporaryFolder WAREHOUSE = new TemporaryFolder();
+
+ private final String catalogName;
+ private final Map<String, String> properties;
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "catalogName = {0}, properties = {1}, isStreaming={2}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ // Create iceberg table in the hadoop catalog and default database.
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop"
+ ),
+ true
+ },
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop"
+ ),
+ false
+ },
+ // Create iceberg table in the hadoop catalog and not_existing_db.
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop",
+ "catalog-database", "not_existing_db"
+ ),
+ true
+ },
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop",
+ "catalog-database", "not_existing_db"
+ ),
+ false
+ },
+ // Create iceberg table in the hive catalog and default database.
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive"
+ ),
+ true
+ },
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive"
+ ),
+ false
+ },
+ // Create iceberg table in the hive catalog and not_existing_db.
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive",
+ "catalog-database", "not_existing_db"
+ ),
+ true
+ },
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive",
+ "catalog-database", "not_existing_db"
+ ),
+ false
+ }
+ );
+ }
+
+ public TestIcebergConnector(String catalogName, Map<String, String> properties, boolean isStreaming) {
+ this.catalogName = catalogName;
+ this.properties = properties;
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() throws TException {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+
+ // Clean the created orphan databases and tables from hive-metastore.
+ if (isHiveCatalog()) {
+ HiveMetaStoreClient metaStoreClient = new HiveMetaStoreClient(hiveConf);
+ try {
+ metaStoreClient.dropTable(databaseName(), TABLE_NAME);
+ if (!isDefaultDatabaseName()) {
+ try {
+ metaStoreClient.dropDatabase(databaseName());
+ } catch (Exception ignored) {
+ // Ignore
+ }
+ }
+ } finally {
+ metaStoreClient.close();
+ }
+ }
+ }
+
+ private void testCreateConnectorTable() {
+ Map<String, String> tableProps = createTableProps();
+
+ // Create table under the flink's current database.
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ Catalog flinkCatalog = factory.createCatalog(catalogName, tableProps, new Configuration());
+ Assert.assertTrue("Should have created the expected database", flinkCatalog.databaseExists(databaseName()));
+ Assert.assertTrue("Should have created the expected table",
+ flinkCatalog.tableExists(new ObjectPath(databaseName(), TABLE_NAME)));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
Review comment:
Is it possible to omit the schema for cases where the underlying table already exists?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659089095
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
Does this discard the rest of the identifier?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] stevenzwu commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r647120788
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
+ .build();
+
+ tEnv = TableEnvironment.create(settings);
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+
+ @Test
+ public void testHadoopCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "local_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
Review comment:
Here we configure the `catalog-name` and `catalog-database` in the props. Flink SQL supports the namespace syntax. trying to understand if that is a problem.
https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/sql/create/
```
CREATE TABLE [catalog_name.][db_name.]table_name
```
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659089095
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
Does this discard the rest of the identifier? What if the database is non-null? It seems like it should not be ignored and is a better way to get the database than a 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] qyw919867774 commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
qyw919867774 commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r838453477
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
+ .build();
+
+ tEnv = TableEnvironment.create(settings);
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+
+ @Test
+ public void testHadoopCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "local_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
Review comment:
understand the meaning of specifying catalog database, but I still don't understand why catalog name must be specified What's the use of specifying catalog name? For Flink, any value of catalog name can be specified, which has no impact on my data processing tasks. @openinx
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659464369
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
Review comment:
I think you are talking about the issue : Is it possible that we access the `catalogTable` by `context.getCatalogTable()` before setting it inside the `context`, if so then the `catalogTable` will be a null object.
I think we don't have to concern this issue because the `catalogTable` was set in the context contructor [here](https://github.com/apache/flink/blob/master/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java#L852). And in the flink code path, the set `catalogTable` must not be null. So I think it's OK 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-882946191
@openinx, I'll take another look at this. Is there a description of the proposed behavior for these tables in the Flink catalog?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] stevenzwu commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r647124382
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,143 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_VALUE)
+ .withDeprecatedKeys("Catalog database");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ options.add(CATALOG_DATABASE);
+ return options;
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return Sets.newHashSet();
}
@Override
public String factoryIdentifier() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return FACTORY_IDENTIFIER;
+ }
+
+ private static TableLoader createTableLoader(CatalogBaseTable catalogTable,
+ Map<String, String> tableProps,
+ String tableName) {
+ Configuration flinkConf = new Configuration();
+ tableProps.forEach(flinkConf::setString);
+
+ String catalogName = flinkConf.getString(CATALOG_NAME);
+ Preconditions.checkNotNull(catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());
+
+ String catalogDatabase = flinkConf.getString(CATALOG_DATABASE);
+ Preconditions.checkNotNull(catalogDatabase, "Table property '%s' cannot be null", CATALOG_DATABASE.key());
+
+ org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
+ CatalogLoader catalogLoader = FlinkCatalogFactory.createCatalogLoader(catalogName, tableProps, hadoopConf);
Review comment:
I am little confused. `FlinkCatalogFactory#createCatalogLoader` is a non-static method. How does this line work?
Also since we are creating a `FlinkCatalogFactory` instance below, we can call the non-static `createCatalogLoader` below
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r646563408
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,150 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ public static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ public static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type.");
+
+ public static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_VALUE)
+ .withDeprecatedKeys("Catalog database");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
Review comment:
That's because the flink's SPI loader will just load from the no-arg constructor, I just want to initialize the `catalog` to be null in this constructor.
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659089939
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ return options;
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_DATABASE);
+ return options;
}
@Override
public String factoryIdentifier() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return FACTORY_IDENTIFIER;
+ }
+
+ private static TableLoader createTableLoader(CatalogBaseTable catalogTable,
+ Map<String, String> tableProps,
+ String tableName) {
+ Configuration flinkConf = new Configuration();
+ tableProps.forEach(flinkConf::setString);
+
+ String catalogName = flinkConf.getString(CATALOG_NAME);
+ Preconditions.checkNotNull(catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());
+
+ String catalogDatabase = flinkConf.getString(CATALOG_DATABASE);
+ Preconditions.checkNotNull(catalogDatabase, "Table property '%s' cannot be null", CATALOG_DATABASE.key());
+
+ org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ FlinkCatalog flinkCatalog = (FlinkCatalog) factory.createCatalog(catalogName, tableProps, hadoopConf);
+ ObjectPath objectPath = new ObjectPath(catalogDatabase, tableName);
+
+ // Create database if not exists in the external catalog.
+ if (!flinkCatalog.databaseExists(catalogDatabase)) {
+ try {
+ flinkCatalog.createDatabase(catalogDatabase, new CatalogDatabaseImpl(Maps.newHashMap(), null), true);
+ } catch (DatabaseAlreadyExistException | CatalogException e) {
+ throw new RuntimeException(String.format("Failed to create database %s.%s", catalogName, catalogDatabase), e);
Review comment:
What about using Iceberg's exceptions here, like [`AlreadyExistsException`](https://github.com/apache/iceberg/blob/master/api/src/main/java/org/apache/iceberg/exceptions/AlreadyExistsException.java), instead of `RuntimeException`?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660553618
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
Review comment:
I got your question, yes, it's possible in your case. For example, when we create a table under the iceberg catalog in flink sql, then it will create the `FlinkDynamicTableFactory` with the specified iceberg catalog.
There should be no problem in that case. the `catalogTable` is the correct flink table which is parsed from SQL such as `INSERT INTO iceberg_catalog.iceberg_db.iceberg_table` , although we don't use it in the current code path.
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
> How do you create an in-memory catalog table pointing to an Iceberg table if the Iceberg table already exists? What if the DDL, like the schema, doesn't match?
If the underlying iceberg table already exists, the we still need to create the in-memory catalog table pointing to it. If the in-memory catalog table schema does not match the underlying iceberg table, then the create table statement won't throw any exception but when executing the `SELECT` query or `INSERT INTO` query it will throw exception if the RowData read from iceberg table could not be parsed by the in-memory catalog table schema. People will need to re-create the in-memory table and map to the underlying table once again. That's the default behavior for flink users, because almost of the flink connectors are the similar behavior ( such as [JDBC](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/jdbc/) connector, [hive](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/hive/hive_read_write/) connector, [hbase](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/hbase/) con
nector).
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
> Why share the table name between the in-memory and external catalog but not the database?
I think you are right. I checked the [jdbc](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/jdbc/) connector, we could specify a different jdbc table name when creating the flink table. It make sense to allow both db & table name to be overridden.
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @ClassRule
+ public static final TemporaryFolder WAREHOUSE = new TemporaryFolder();
+
+ private final String catalogName;
+ private final Map<String, String> properties;
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "catalogName = {0}, properties = {1}, isStreaming={2}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ // Create iceberg table in the hadoop catalog and default database.
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop"
+ ),
+ true
+ },
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop"
+ ),
+ false
+ },
+ // Create iceberg table in the hadoop catalog and not_existing_db.
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop",
+ "catalog-database", "not_existing_db"
+ ),
+ true
+ },
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop",
+ "catalog-database", "not_existing_db"
+ ),
+ false
+ },
+ // Create iceberg table in the hive catalog and default database.
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive"
+ ),
+ true
+ },
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive"
+ ),
+ false
+ },
+ // Create iceberg table in the hive catalog and not_existing_db.
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive",
+ "catalog-database", "not_existing_db"
+ ),
+ true
+ },
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive",
+ "catalog-database", "not_existing_db"
+ ),
+ false
+ }
+ );
+ }
+
+ public TestIcebergConnector(String catalogName, Map<String, String> properties, boolean isStreaming) {
+ this.catalogName = catalogName;
+ this.properties = properties;
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() throws TException {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+
+ // Clean the created orphan databases and tables from hive-metastore.
+ if (isHiveCatalog()) {
+ HiveMetaStoreClient metaStoreClient = new HiveMetaStoreClient(hiveConf);
+ try {
+ metaStoreClient.dropTable(databaseName(), TABLE_NAME);
+ if (!isDefaultDatabaseName()) {
+ try {
+ metaStoreClient.dropDatabase(databaseName());
+ } catch (Exception ignored) {
+ // Ignore
+ }
+ }
+ } finally {
+ metaStoreClient.close();
+ }
+ }
+ }
+
+ private void testCreateConnectorTable() {
+ Map<String, String> tableProps = createTableProps();
+
+ // Create table under the flink's current database.
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ Catalog flinkCatalog = factory.createCatalog(catalogName, tableProps, new Configuration());
+ Assert.assertTrue("Should have created the expected database", flinkCatalog.databaseExists(databaseName()));
+ Assert.assertTrue("Should have created the expected table",
+ flinkCatalog.tableExists(new ObjectPath(databaseName(), TABLE_NAME)));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
Review comment:
Unfortunately, the flink sql don't provide the syntax to support it now, although I agree it's a better user-experience . Currently, the flink sql provide the [CREATE TABLE a LIKE b](https://iceberg.apache.org/flink/#create-table-like), but it still require that the table b is a flink catalog 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] qyw919867774 commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
qyw919867774 commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r838453477
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
+ .build();
+
+ tEnv = TableEnvironment.create(settings);
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+
+ @Test
+ public void testHadoopCatalog() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "local_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE hadoop_table (id BIGINT, data STRING) WITH %s", toWithClause(tableProps));
Review comment:
I understand the meaning of specifying catalog database, but I still don't understand why catalog name must be specified What's the use of specifying catalog name? For Flink, any value of catalog name can be specified, which has no impact on my data processing tasks. @openinx
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660202805
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @ClassRule
+ public static final TemporaryFolder WAREHOUSE = new TemporaryFolder();
+
+ private final String catalogName;
+ private final Map<String, String> properties;
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "catalogName = {0}, properties = {1}, isStreaming={2}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ // Create iceberg table in the hadoop catalog and default database.
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop"
+ ),
+ true
+ },
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop"
+ ),
+ false
+ },
+ // Create iceberg table in the hadoop catalog and not_existing_db.
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop",
+ "catalog-database", "not_existing_db"
+ ),
+ true
+ },
+ new Object[] {
+ "testhadoop",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hadoop",
+ "catalog-database", "not_existing_db"
+ ),
+ false
+ },
+ // Create iceberg table in the hive catalog and default database.
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive"
+ ),
+ true
+ },
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive"
+ ),
+ false
+ },
+ // Create iceberg table in the hive catalog and not_existing_db.
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive",
+ "catalog-database", "not_existing_db"
+ ),
+ true
+ },
+ new Object[] {
+ "testhive",
+ ImmutableMap.of(
+ "connector", "iceberg",
+ "catalog-type", "hive",
+ "catalog-database", "not_existing_db"
+ ),
+ false
+ }
+ );
+ }
+
+ public TestIcebergConnector(String catalogName, Map<String, String> properties, boolean isStreaming) {
+ this.catalogName = catalogName;
+ this.properties = properties;
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() throws TException {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+
+ // Clean the created orphan databases and tables from hive-metastore.
+ if (isHiveCatalog()) {
+ HiveMetaStoreClient metaStoreClient = new HiveMetaStoreClient(hiveConf);
+ try {
+ metaStoreClient.dropTable(databaseName(), TABLE_NAME);
+ if (!isDefaultDatabaseName()) {
+ try {
+ metaStoreClient.dropDatabase(databaseName());
+ } catch (Exception ignored) {
+ // Ignore
+ }
+ }
+ } finally {
+ metaStoreClient.close();
+ }
+ }
+ }
+
+ private void testCreateConnectorTable() {
+ Map<String, String> tableProps = createTableProps();
+
+ // Create table under the flink's current database.
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ Catalog flinkCatalog = factory.createCatalog(catalogName, tableProps, new Configuration());
+ Assert.assertTrue("Should have created the expected database", flinkCatalog.databaseExists(databaseName()));
+ Assert.assertTrue("Should have created the expected table",
+ flinkCatalog.tableExists(new ObjectPath(databaseName(), TABLE_NAME)));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
Review comment:
Is it possible to omit the schema for cases where the underlying table already exists?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-882946191
@openinx, I'll take another look at this. Is there a description of the proposed behavior for these tables in the Flink catalog?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-911159322
I think this PR has been fully reviewed. Seeem this useful feature has been blocked for so long, I plan to get this merged into the offical apache iceberg repo. If people have more comments or improvements, pls file new issue or PR to address the following thing. Thanks all 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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 merged pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx merged pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r657961655
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "isStreaming={0}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ new Object[] {true},
+ new Object[] {false}
+ );
+ }
+
+ public TestIcebergConnector(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+ }
+
+ @Test
+ public void testHadoop() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
Review comment:
> drop table doesn't purge data and metadata files, right?
Yes, this `DROP TABLE test_table` is actually doing `DROP TABLE default_catalog.default_database.test_table`, which means dropping the `test_table` from flink's in-memory catalog, the underlying iceberg catalog & table (`test-hadoop.default.test_table`) that flink table is mapping to won't be effected. Here I want to make sure that we could map the flink in-memory table to the underlying iceberg table again once we drop and create again.
--
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 pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-867636720
> @openinx, sorry for the delay. I'll make some time to review this.
>
> I also know that while I was out there were a few PRs that I wasn't able to help move along. If you have a list of PRs that are important for you to get in, please send them to me and I'll make time to get them reviewed as I can. Thank you for being patient with me!
@rdblue , thanks for your time. All the PRs that were published by me are here : https://github.com/apache/iceberg/pulls/openinx . There are three parts :
1. The first part is about improvement for iceberg+flink module, I think this PR is the highest priority. Others are related to flink CDC write path, let's discuss them in the second parts.
2. Write & analysis the change log events in format v2 iceberg table. After #2410, we could ingest change log events into apache iceberg by pure flink SQL now. But we still have other important issues that need to be addressed:
a. How to ensure the stability of streaming jobs ? Currently, the most important PR is : https://github.com/apache/iceberg/pull/2680 . There are some other problems. For example, random change log events may cause a large number of parquet writers to be opened in a checkpoint, and eventually cause OOM of the job. I will open a specific issue for that.
b. How to implement the compaction action for v2. As we know we are implementing the minor compaction (I mean translate the equality deletes to pos-deletes ? ) , but I think we may could get this [major compaction](https://github.com/apache/iceberg/pull/2303) merged first (Of course , the current patch still has few concerns that need to be addressed, I will do) . I also had a temporary fix for https://github.com/apache/iceberg/issues/2308, and will publish it to the apache repo for reviewing.
3. Aliyun OSS + DLF integration work. I had an opened PR for [aliyun OSS](https://github.com/apache/iceberg/pull/2230) and an pending PR for[ aliyun DLF integration](https://github.com/openinx/incubator-iceberg/pull/5) (now it's in my personal repo) . Let's put these PRs aside for now, and I will try to split these PRs appropriately so that we can better review them in the future.
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659467596
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ return options;
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_DATABASE);
+ return options;
}
@Override
public String factoryIdentifier() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return FACTORY_IDENTIFIER;
+ }
+
+ private static TableLoader createTableLoader(CatalogBaseTable catalogTable,
+ Map<String, String> tableProps,
+ String tableName) {
+ Configuration flinkConf = new Configuration();
+ tableProps.forEach(flinkConf::setString);
+
+ String catalogName = flinkConf.getString(CATALOG_NAME);
+ Preconditions.checkNotNull(catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());
+
+ String catalogDatabase = flinkConf.getString(CATALOG_DATABASE);
+ Preconditions.checkNotNull(catalogDatabase, "Table property '%s' cannot be null", CATALOG_DATABASE.key());
+
+ org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ FlinkCatalog flinkCatalog = (FlinkCatalog) factory.createCatalog(catalogName, tableProps, hadoopConf);
+ ObjectPath objectPath = new ObjectPath(catalogDatabase, tableName);
+
+ // Create database if not exists in the external catalog.
+ if (!flinkCatalog.databaseExists(catalogDatabase)) {
+ try {
+ flinkCatalog.createDatabase(catalogDatabase, new CatalogDatabaseImpl(Maps.newHashMap(), null), true);
+ } catch (DatabaseAlreadyExistException | CatalogException e) {
+ throw new RuntimeException(String.format("Failed to create database %s.%s", catalogName, catalogDatabase), e);
Review comment:
Yeah, that' looks pretty good to me !
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660202128
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
Review comment:
Sorry, I meant is there a case when `context.getCatalogTable()` is non-null but passed to a source where `catalog` is set in the constructor?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r657967579
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "isStreaming={0}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ new Object[] {true},
+ new Object[] {false}
+ );
+ }
+
+ public TestIcebergConnector(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+ }
+
+ @Test
+ public void testHadoop() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+ }
+
+ @Test
+ public void testHadoopCreateDatabaseIfNotExist() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "not_existing_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ Catalog flinkCatalog = factory.createCatalog("test-hadoop", tableProps, new Configuration());
+ Assert.assertTrue("Should have created the database",
+ flinkCatalog.databaseExists("not_existing_db"));
+ Assert.assertTrue("Should have created the table",
+ flinkCatalog.tableExists(new ObjectPath("not_existing_db", TABLE_NAME)));
+ }
+
+ @Test
+ public void testHive() throws Exception {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hive");
+ tableProps.put("catalog-type", "hive");
+ tableProps.put(CatalogProperties.URI, FlinkCatalogTestBase.getURI(hiveConf));
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE %s(id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')", TABLE_NAME);
+ Assert.assertEquals("Should have the expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ sql("DROP TABLE %s", TABLE_NAME);
+ HiveMetaStoreClient metaStoreClient = new HiveMetaStoreClient(hiveConf);
Review comment:
Moving the cleanup block in the try-finally looks great to me. About splitting it into two files for hadoop and hive, I think we won't have too much test branches for hadoop and hive because most of the flink+iceberg integration test work will be accomplished in the flink catalog test suite, this UT is only used for making the `CREATE TABLE xx () WITH ('connector'='iceberg')` work, so I made them into this whole test 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] gaborgsomogyi commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r655146166
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static TableEnvironment tEnv;
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+
+ @BeforeClass
+ public static void beforeClass() {
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inBatchMode()
Review comment:
OK, this is nice to have.
--
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] stevenzwu commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r657513503
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ return options;
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_DATABASE);
+ return options;
}
@Override
public String factoryIdentifier() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return FACTORY_IDENTIFIER;
+ }
+
+ private static TableLoader createTableLoader(CatalogBaseTable catalogTable,
+ Map<String, String> tableProps,
+ String tableName) {
+ Configuration flinkConf = new Configuration();
+ tableProps.forEach(flinkConf::setString);
+
+ String catalogName = flinkConf.getString(CATALOG_NAME);
+ Preconditions.checkNotNull(catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());
+
+ String catalogDatabase = flinkConf.getString(CATALOG_DATABASE);
+ Preconditions.checkNotNull(catalogDatabase, "Table property '%s' cannot be null", CATALOG_DATABASE.key());
+
+ org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ FlinkCatalog flinkCatalog = (FlinkCatalog) factory.createCatalog(catalogName, tableProps, hadoopConf);
+ ObjectPath objectPath = new ObjectPath(catalogDatabase, tableName);
+
+ // Create database if not exists in the external catalog.
+ if (!flinkCatalog.databaseExists(catalogDatabase)) {
+ try {
+ flinkCatalog.createDatabase(catalogDatabase, new CatalogDatabaseImpl(Maps.newHashMap(), null), true);
+ } catch (DatabaseAlreadyExistException | CatalogException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ // Create table if not exists in the external catalog.
+ if (!flinkCatalog.tableExists(objectPath)) {
+ try {
+ flinkCatalog.createTable(objectPath, catalogTable, true);
+ } catch (TableAlreadyExistException | CatalogException e) {
+ throw new RuntimeException(e);
Review comment:
nit: add an error msg
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659089571
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
Review comment:
When is `catalogTable` set in the context? Is it possible that it is set and the factory's `context` is non-null? In that case, should an error be thrown?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659090597
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "isStreaming={0}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ new Object[] {true},
+ new Object[] {false}
+ );
+ }
+
+ public TestIcebergConnector(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+ }
+
+ @Test
+ public void testHadoop() {
Review comment:
I would like to have tests for when the database option and the one from the identifier conflict.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] stevenzwu commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r657518290
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "isStreaming={0}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ new Object[] {true},
+ new Object[] {false}
+ );
+ }
+
+ public TestIcebergConnector(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+ }
+
+ @Test
+ public void testHadoop() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+ }
+
+ @Test
+ public void testHadoopCreateDatabaseIfNotExist() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put("catalog-database", "not_existing_db");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ Catalog flinkCatalog = factory.createCatalog("test-hadoop", tableProps, new Configuration());
+ Assert.assertTrue("Should have created the database",
+ flinkCatalog.databaseExists("not_existing_db"));
+ Assert.assertTrue("Should have created the table",
+ flinkCatalog.tableExists(new ObjectPath("not_existing_db", TABLE_NAME)));
+ }
+
+ @Test
+ public void testHive() throws Exception {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hive");
+ tableProps.put("catalog-type", "hive");
+ tableProps.put(CatalogProperties.URI, FlinkCatalogTestBase.getURI(hiveConf));
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE %s(id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC')", TABLE_NAME);
+ Assert.assertEquals("Should have the expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "AAA"), Row.of(2L, "BBB"), Row.of(3L, "CCC")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ sql("DROP TABLE %s", TABLE_NAME);
+ HiveMetaStoreClient metaStoreClient = new HiveMetaStoreClient(hiveConf);
Review comment:
should we run this cleanup block in the try-finally? Another option to split this file into two files (one for hadoop and one for hive). we can also potentially remove some of the redundant code with a base test 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] openinx commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r660562573
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
Review comment:
> How do you create an in-memory catalog table pointing to an Iceberg table if the Iceberg table already exists? What if the DDL, like the schema, doesn't match?
If the underlying iceberg table already exists, the we still need to create the in-memory catalog table pointing to it. If the in-memory catalog table schema does not match the underlying iceberg table, then the create table statement won't throw any exception but when executing the `SELECT` query or `INSERT INTO` query it will throw exception if the RowData read from iceberg table could not be parsed by the in-memory catalog table schema. People will need to re-create the in-memory table and map to the underlying table once again. That's the default behavior for flink users, because almost of the flink connectors are the similar behavior ( such as [JDBC](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/jdbc/) connector, [hive](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/hive/hive_read_write/) connector, [hbase](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/hbase/) con
nector).
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] stevenzwu commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r645666945
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,150 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ public static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ public static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type.");
+
+ public static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_VALUE)
+ .withDeprecatedKeys("Catalog database");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
Review comment:
it wasn't immediately clear to me why we added the default constructor
--
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 pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-911156610
> Is there a description of the proposed behavior for these tables in the Flink catalog?
Yes, the default in-memory flink catalog will maintain those tables inside it, once people drop those tables from the in-memory flink catalog. The in-memory flink catalog will don't see the tables, but the underlying iceberg tables in file system are still there. If people want to drop those data from tables, the correct way is to drop tables from the backend catalog.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] stevenzwu commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r657513398
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ return options;
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_DATABASE);
+ return options;
}
@Override
public String factoryIdentifier() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return FACTORY_IDENTIFIER;
+ }
+
+ private static TableLoader createTableLoader(CatalogBaseTable catalogTable,
+ Map<String, String> tableProps,
+ String tableName) {
+ Configuration flinkConf = new Configuration();
+ tableProps.forEach(flinkConf::setString);
+
+ String catalogName = flinkConf.getString(CATALOG_NAME);
+ Preconditions.checkNotNull(catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());
+
+ String catalogDatabase = flinkConf.getString(CATALOG_DATABASE);
+ Preconditions.checkNotNull(catalogDatabase, "Table property '%s' cannot be null", CATALOG_DATABASE.key());
+
+ org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ FlinkCatalog flinkCatalog = (FlinkCatalog) factory.createCatalog(catalogName, tableProps, hadoopConf);
+ ObjectPath objectPath = new ObjectPath(catalogDatabase, tableName);
+
+ // Create database if not exists in the external catalog.
+ if (!flinkCatalog.databaseExists(catalogDatabase)) {
+ try {
+ flinkCatalog.createDatabase(catalogDatabase, new CatalogDatabaseImpl(Maps.newHashMap(), null), true);
+ } catch (DatabaseAlreadyExistException | CatalogException e) {
+ throw new RuntimeException(e);
Review comment:
nit: maybe add an error msg like "Failed to create database"
--
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 pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-866473189
@rdblue , @stevenzwu do you have another other concerns for this PR ? I think this is a very important issue for flink users because it's more in line with the usage habits of flink users.
--
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 #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r659090021
##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
##########
@@ -19,56 +19,155 @@
package org.apache.iceberg.flink;
+import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.utils.TableSchemaUtils;
+import org.apache.flink.util.Preconditions;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
public class FlinkDynamicTableFactory implements DynamicTableSinkFactory, DynamicTableSourceFactory {
+ private static final String FACTORY_IDENTIFIER = "iceberg";
+
+ private static final ConfigOption<String> CATALOG_NAME =
+ ConfigOptions.key("catalog-name")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog name");
+
+ private static final ConfigOption<String> CATALOG_TYPE =
+ ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+ private static final ConfigOption<String> CATALOG_DATABASE =
+ ConfigOptions.key("catalog-database")
+ .stringType()
+ .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+ .withDescription("Database name that managed in the iceberg catalog.");
+
private final FlinkCatalog catalog;
+ public FlinkDynamicTableFactory() {
+ this.catalog = null;
+ }
+
public FlinkDynamicTableFactory(FlinkCatalog catalog) {
this.catalog = catalog;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ ObjectIdentifier objectIdentifier = context.getObjectIdentifier();
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
- return new IcebergTableSource(tableLoader, tableSchema, context.getCatalogTable().getOptions(),
- context.getConfiguration());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectIdentifier.toObjectPath());
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectIdentifier.getObjectName());
+ }
+
+ return new IcebergTableSource(tableLoader, tableSchema, tableProps, context.getConfiguration());
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
ObjectPath objectPath = context.getObjectIdentifier().toObjectPath();
- TableLoader tableLoader = createTableLoader(objectPath);
+ Map<String, String> tableProps = context.getCatalogTable().getOptions();
+ CatalogTable catalogTable = context.getCatalogTable();
TableSchema tableSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
+
+ TableLoader tableLoader;
+ if (catalog != null) {
+ tableLoader = createTableLoader(catalog, objectPath);
+ } else {
+ tableLoader = createTableLoader(catalogTable, tableProps, objectPath.getObjectName());
+ }
+
return new IcebergTableSink(tableLoader, tableSchema);
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_TYPE);
+ options.add(CATALOG_NAME);
+ return options;
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ Set<ConfigOption<?>> options = Sets.newHashSet();
+ options.add(CATALOG_DATABASE);
+ return options;
}
@Override
public String factoryIdentifier() {
- throw new UnsupportedOperationException("Iceberg Table Factory can not be loaded from Java SPI");
+ return FACTORY_IDENTIFIER;
+ }
+
+ private static TableLoader createTableLoader(CatalogBaseTable catalogTable,
+ Map<String, String> tableProps,
+ String tableName) {
+ Configuration flinkConf = new Configuration();
+ tableProps.forEach(flinkConf::setString);
+
+ String catalogName = flinkConf.getString(CATALOG_NAME);
+ Preconditions.checkNotNull(catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());
+
+ String catalogDatabase = flinkConf.getString(CATALOG_DATABASE);
+ Preconditions.checkNotNull(catalogDatabase, "Table property '%s' cannot be null", CATALOG_DATABASE.key());
+
+ org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
+ FlinkCatalogFactory factory = new FlinkCatalogFactory();
+ FlinkCatalog flinkCatalog = (FlinkCatalog) factory.createCatalog(catalogName, tableProps, hadoopConf);
+ ObjectPath objectPath = new ObjectPath(catalogDatabase, tableName);
+
+ // Create database if not exists in the external catalog.
+ if (!flinkCatalog.databaseExists(catalogDatabase)) {
+ try {
+ flinkCatalog.createDatabase(catalogDatabase, new CatalogDatabaseImpl(Maps.newHashMap(), null), true);
+ } catch (DatabaseAlreadyExistException | CatalogException e) {
+ throw new RuntimeException(String.format("Failed to create database %s.%s", catalogName, catalogDatabase), e);
+ }
+ }
+
+ // Create table if not exists in the external catalog.
+ if (!flinkCatalog.tableExists(objectPath)) {
+ try {
+ flinkCatalog.createTable(objectPath, catalogTable, true);
+ } catch (TableAlreadyExistException | CatalogException e) {
+ throw new RuntimeException(String.format("Failed to create table %s.%s", catalogName, objectPath), e);
Review comment:
Same here. I think it is better to throw Iceberg exceptions if they exist.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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] stevenzwu commented on a change in pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
stevenzwu commented on a change in pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#discussion_r657516562
##########
File path: flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+
+package org.apache.iceberg.flink;
+
+import java.util.Map;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public class TestIcebergConnector extends FlinkTestBase {
+
+ private static final String TABLE_NAME = "test_table";
+
+ @Rule
+ public final TemporaryFolder warehouse = new TemporaryFolder();
+
+ private final boolean isStreaming;
+ private volatile TableEnvironment tEnv;
+
+ @Parameterized.Parameters(name = "isStreaming={0}")
+ public static Iterable<Object[]> parameters() {
+ return Lists.newArrayList(
+ new Object[] {true},
+ new Object[] {false}
+ );
+ }
+
+ public TestIcebergConnector(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ if (tEnv == null) {
+ synchronized (this) {
+ if (tEnv == null) {
+ EnvironmentSettings.Builder settingsBuilder = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner();
+ if (isStreaming) {
+ settingsBuilder.inStreamingMode();
+ StreamExecutionEnvironment env = StreamExecutionEnvironment
+ .getExecutionEnvironment(MiniClusterResource.DISABLE_CLASSLOADER_CHECK_CONFIG);
+ env.enableCheckpointing(400);
+ env.setMaxParallelism(2);
+ env.setParallelism(2);
+ tEnv = StreamTableEnvironment.create(env, settingsBuilder.build());
+ } else {
+ settingsBuilder.inBatchMode();
+ tEnv = TableEnvironment.create(settingsBuilder.build());
+ }
+ // Set only one parallelism.
+ tEnv.getConfig().getConfiguration()
+ .set(CoreOptions.DEFAULT_PARALLELISM, 1)
+ .set(FlinkTableOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+ }
+ }
+ }
+ return tEnv;
+ }
+
+ @After
+ public void after() {
+ sql("DROP TABLE IF EXISTS %s", TABLE_NAME);
+ }
+
+ @Test
+ public void testHadoop() {
+ Map<String, String> tableProps = Maps.newHashMap();
+ tableProps.put("connector", "iceberg");
+ tableProps.put("catalog-name", "test-hadoop");
+ tableProps.put("catalog-type", "hadoop");
+ tableProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseRoot());
+
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ sql("INSERT INTO %s VALUES (1, 'aaa'), (2, 'bbb'), (3, 'ccc')", TABLE_NAME);
+ Assert.assertEquals("Should have expected rows",
+ Lists.newArrayList(Row.of(1L, "aaa"), Row.of(2L, "bbb"), Row.of(3L, "ccc")),
+ sql("SELECT * FROM %s", TABLE_NAME));
+
+ // Drop and create it again.
+ sql("DROP TABLE %s", TABLE_NAME);
+ sql("CREATE TABLE %s (id BIGINT, data STRING) WITH %s", TABLE_NAME, toWithClause(tableProps));
+ Assert.assertEquals("Should have expected rows",
Review comment:
drop table doesn't purge data and metadata files, right? table creation again will be able to see the old data. I guess it is to prevent the disaster from accidental "drop 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] rdblue commented on pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-868893475
@openinx, I had a few questions but this looks close overall.
Thanks for giving me an overview of what you're currently working on. I agree that the v2 writes are important. And I think we need the initial fix for #2308 as soon as possible. I'll mark that as a blocker for 0.12.0.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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 pull request #2666: Flink: Support create iceberg table with 'connector'='iceberg'
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #2666:
URL: https://github.com/apache/iceberg/pull/2666#issuecomment-878701949
Ping @rdblue , How is your feeling about this PR now ?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
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