luckychacha opened a new issue #4398:
URL: https://github.com/apache/iceberg/issues/4398
Hello, I meet an exception when I execute "INSERT" or "SELECT" on a table.
Here is my code:
``` java
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
public class Insert {
public static void main(String[] args) {
EnvironmentSettings settings = EnvironmentSettings
.newInstance()
.inStreamingMode()
.build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
TableResult tableResult = tableEnv.executeSql(
"CREATE CATALOG hadoop_catalog WITH (\n" +
" 'type'='iceberg',\n" +
" 'catalog-type'='hadoop',\n" +
"
'warehouse'='hdfs://localhost:8020/tmp/iceberg/warehouse',\n" +
" 'property-version'='1'\n" +
")");
tableEnv.useCatalog("hadoop_catalog");
tableEnv.executeSql("CREATE DATABASE iceberg_hadoop_db");
tableEnv.useDatabase("iceberg_hadoop_db");
tableEnv.executeSql("CREATE TABLE
`hadoop_catalog`.`iceberg_hadoop_db`.`iceberg_001` (\n" +
" id BIGINT COMMENT 'unique id',\n" +
" data STRING\n" +
")");
tableEnv.executeSql("INSERT INTO
`hadoop_catalog`.`iceberg_hadoop_db`.`iceberg_001` VALUES (106,'106')");
}
}
```
Exception is:
```
Exception in thread "main" org.apache.flink.table.api.ValidationException:
Unable to create a sink for writing table
'hadoop_catalog.iceberg_hadoop_db.iceberg_001'.
Table options are:
at
org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:184)
at
org.apache.flink.table.planner.delegation.PlannerBase.getTableSink(PlannerBase.scala:394)
at
org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:222)
at
org.apache.flink.table.planner.delegation.PlannerBase.$anonfun$translate$1(PlannerBase.scala:182)
at
scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
at scala.collection.Iterator.foreach(Iterator.scala:937)
at scala.collection.Iterator.foreach$(Iterator.scala:937)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
at scala.collection.IterableLike.foreach(IterableLike.scala:70)
at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike.map(TraversableLike.scala:233)
at scala.collection.TraversableLike.map$(TraversableLike.scala:226)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at
org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:182)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1665)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:752)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:872)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:742)
at Insert.main(Insert.java:30)
Caused by: java.lang.NoSuchMethodError:
org.apache.flink.table.factories.DynamicTableFactory$Context.getCatalogTable()Lorg/apache/flink/table/catalog/CatalogTable;
at
org.apache.iceberg.flink.FlinkDynamicTableFactory.createDynamicTableSink(FlinkDynamicTableFactory.java:52)
at
org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:181)
... 19 more
```
iceberg version and flink version is:
```
<flink.version>1.14.4</flink.version>
<iceberg.version>0.12.1</iceberg.version>
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
