Unable to access apache ozone using iceberg catalog.

[cloudcheflabs]

Hi,

I have run spark job to insert data to iceberg table on apache ozone using hive metastore. It has created iceberg schema and saved data as iceberg table successfully. I could also read iceberg table on ozone using spark connected to hive metastore without problem.

But, when I tried to read iceberg table from ozone using iceberg catalog connected to hive metastore, it failed with exceptions.
My iceberg catalog looks like this.

connector.name=iceberg
hive.metastore.uri=thrift://<hive-metastore-url>
hive.config.resources=<ozone-conf-dir>/hadoop/core-site.xml,<ozone-conf-dir>/hadoop/ozone-site.xml

I have added ozone-filesystem-hadoop3-1.3.0.jar to <trino-home>/plugin/iceberg, and then I tried, but it failed.

java.lang.NoClassDefFoundError: com/google/protobuf/ServiceException
	at org.apache.hadoop.ozone.om.protocolPB.Hadoop3OmTransportFactory.createOmTransport(Hadoop3OmTransportFactory.java:33)
	at org.apache.hadoop.ozone.om.protocolPB.OmTransportFactory.create(OmTransportFactory.java:42)
	at org.apache.hadoop.ozone.client.rpc.RpcClient.createOmTransport(RpcClient.java:400)
	at org.apache.hadoop.ozone.client.rpc.RpcClient.<init>(RpcClient.java:229)
	at org.apache.hadoop.ozone.client.OzoneClientFactory.getClientProtocol(OzoneClientFactory.java:247)
	at org.apache.hadoop.ozone.client.OzoneClientFactory.getRpcClient(OzoneClientFactory.java:114)
	at org.apache.hadoop.fs.ozone.BasicRootedOzoneClientAdapterImpl.<init>(BasicRootedOzoneClientAdapterImpl.java:182)
	at org.apache.hadoop.fs.ozone.RootedOzoneClientAdapterImpl.<init>(RootedOzoneClientAdapterImpl.java:51)
	at org.apache.hadoop.fs.ozone.RootedOzoneFileSystem.createAdapter(RootedOzoneFileSystem.java:93)
	at org.apache.hadoop.fs.ozone.BasicRootedOzoneFileSystem.initialize(BasicRootedOzoneFileSystem.java:150)
	at io.trino.hdfs.TrinoFileSystemCache.createFileSystem(TrinoFileSystemCache.java:155)
	at io.trino.hdfs.TrinoFileSystemCache$FileSystemHolder.createFileSystemOnce(TrinoFileSystemCache.java:293)
	at io.trino.hdfs.TrinoFileSystemCache.getInternal(TrinoFileSystemCache.java:135)
	at io.trino.hdfs.TrinoFileSystemCache.get(TrinoFileSystemCache.java:91)
	at org.apache.hadoop.fs.ForwardingFileSystemCache.get(ForwardingFileSystemCache.java:38)
	at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:479)
	at org.apache.hadoop.fs.Path.getFileSystem(Path.java:361)
	at io.trino.hdfs.HdfsEnvironment.lambda$getFileSystem$0(HdfsEnvironment.java:104)
	at io.trino.hdfs.authentication.NoHdfsAuthentication.doAs(NoHdfsAuthentication.java:25)
	at io.trino.hdfs.HdfsEnvironment.getFileSystem(HdfsEnvironment.java:103)
	at io.trino.hdfs.HdfsEnvironment.getFileSystem(HdfsEnvironment.java:97)
	at io.trino.filesystem.hdfs.HdfsInputFile.openFile(HdfsInputFile.java:113)
	at io.trino.filesystem.hdfs.HdfsInputFile.newStream(HdfsInputFile.java:69)
	at io.trino.filesystem.tracing.Tracing.withTracing(Tracing.java:47)
	at io.trino.filesystem.tracing.TracingInputFile.newStream(TracingInputFile.java:64)
	at io.trino.plugin.iceberg.fileio.ForwardingInputFile.newStream(ForwardingInputFile.java:52)
	at org.apache.iceberg.TableMetadataParser.read(TableMetadataParser.java:272)
	at io.trino.plugin.iceberg.catalog.AbstractIcebergTableOperations.lambda$refreshFromMetadataLocation$1(AbstractIcebergTableOperations.java:240)
	at dev.failsafe.Functions.lambda$toCtxSupplier$11(Functions.java:243)
	at dev.failsafe.Functions.lambda$get$0(Functions.java:46)
	at dev.failsafe.internal.RetryPolicyExecutor.lambda$apply$0(RetryPolicyExecutor.java:74)
	at dev.failsafe.SyncExecutionImpl.executeSync(SyncExecutionImpl.java:187)
	at dev.failsafe.FailsafeExecutor.call(FailsafeExecutor.java:376)
	at dev.failsafe.FailsafeExecutor.get(FailsafeExecutor.java:112)
	at io.trino.plugin.iceberg.catalog.AbstractIcebergTableOperations.refreshFromMetadataLocation(AbstractIcebergTableOperations.java:240)
	at io.trino.plugin.iceberg.catalog.AbstractIcebergTableOperations.refresh(AbstractIcebergTableOperations.java:134)
	at io.trino.plugin.iceberg.catalog.AbstractIcebergTableOperations.current(AbstractIcebergTableOperations.java:117)
	at io.trino.plugin.iceberg.catalog.hms.TrinoHiveCatalog.lambda$loadTable$11(TrinoHiveCatalog.java:374)
	at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1708)
	at io.trino.plugin.iceberg.catalog.hms.TrinoHiveCatalog.loadTable(TrinoHiveCatalog.java:372)
	at io.trino.plugin.iceberg.IcebergMetadata.getTableHandle(IcebergMetadata.java:371)
	at io.trino.plugin.base.classloader.ClassLoaderSafeConnectorMetadata.getTableHandle(ClassLoaderSafeConnectorMetadata.java:1085)
	at io.trino.tracing.TracingConnectorMetadata.getTableHandle(TracingConnectorMetadata.java:143)
	at io.trino.metadata.MetadataManager.lambda$getTableHandle$5(MetadataManager.java:283)
	at java.base/java.util.Optional.flatMap(Optional.java:289)
	at io.trino.metadata.MetadataManager.getTableHandle(MetadataManager.java:277)
	at io.trino.metadata.MetadataManager.getRedirectionAwareTableHandle(MetadataManager.java:1572)
	at io.trino.metadata.MetadataManager.getRedirectionAwareTableHandle(MetadataManager.java:1564)
	at io.trino.tracing.TracingMetadata.getRedirectionAwareTableHandle(TracingMetadata.java:1272)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.getTableHandle(StatementAnalyzer.java:5411)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitTable(StatementAnalyzer.java:2183)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitTable(StatementAnalyzer.java:485)
	at io.trino.sql.tree.Table.accept(Table.java:60)
	at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:502)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.analyzeFrom(StatementAnalyzer.java:4470)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuerySpecification(StatementAnalyzer.java:2947)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuerySpecification(StatementAnalyzer.java:485)
	at io.trino.sql.tree.QuerySpecification.accept(QuerySpecification.java:155)
	at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:502)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:510)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuery(StatementAnalyzer.java:1467)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuery(StatementAnalyzer.java:485)
	at io.trino.sql.tree.Query.accept(Query.java:107)
	at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:502)
	at io.trino.sql.analyzer.StatementAnalyzer.analyze(StatementAnalyzer.java:464)
	at io.trino.sql.analyzer.Analyzer.analyze(Analyzer.java:97)
	at io.trino.sql.analyzer.Analyzer.analyze(Analyzer.java:86)
	at io.trino.execution.SqlQueryExecution.analyze(SqlQueryExecution.java:271)
	at io.trino.execution.SqlQueryExecution.<init>(SqlQueryExecution.java:206)
	at io.trino.execution.SqlQueryExecution$SqlQueryExecutionFactory.createQueryExecution(SqlQueryExecution.java:845)
	at io.trino.dispatcher.LocalDispatchQueryFactory.lambda$createDispatchQuery$0(LocalDispatchQueryFactory.java:153)
	at io.trino.$gen.Trino_419____20230623_142526_2.call(Unknown Source)
	at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131)
	at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:74)
	at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)

When protobuf-java-xxx.jar added to plugin/iceberg directory, I got another exceptions.

2023-06-23T23:56:25.263+0900	ERROR	Query-20230623_145051_00000_dhj5w-159	org.apache.hadoop.ozone.om.ha.OMFailoverProxyProviderBase	Failed to connect to OMs: [om1, om2, om3]. Attempted 500 failovers.
2023-06-23T23:56:25.263+0900	ERROR	Query-20230623_145051_00000_dhj5w-159	org.apache.hadoop.ozone.client.OzoneClientFactory	Couldn't create RpcClient protocol exception: 
java.lang.IllegalStateException
	at io.trino.hadoop.$internal.com.google.common.base.Preconditions.checkState(Preconditions.java:129)
	at org.apache.hadoop.ipc.Client.setCallIdAndRetryCount(Client.java:118)
	at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invokeMethod(RetryInvocationHandler.java:162)
	at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invoke(RetryInvocationHandler.java:157)
	at org.apache.hadoop.io.retry.RetryInvocationHandler$Call.invokeOnce(RetryInvocationHandler.java:95)
	at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:359)
	at jdk.proxy5/jdk.proxy5.$Proxy265.submitRequest(Unknown Source)
	at org.apache.hadoop.ozone.om.protocolPB.Hadoop3OmTransport.submitRequest(Hadoop3OmTransport.java:80)
	at org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolClientSideTranslatorPB.submitRequest(OzoneManagerProtocolClientSideTranslatorPB.java:289)
	at org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolClientSideTranslatorPB.getServiceInfo(OzoneManagerProtocolClientSideTranslatorPB.java:1454)
	at org.apache.hadoop.ozone.client.rpc.RpcClient.<init>(RpcClient.java:239)
	at org.apache.hadoop.ozone.client.OzoneClientFactory.getClientProtocol(OzoneClientFactory.java:247)
	at org.apache.hadoop.ozone.client.OzoneClientFactory.getRpcClient(OzoneClientFactory.java:114)
	at org.apache.hadoop.fs.ozone.BasicRootedOzoneClientAdapterImpl.<init>(BasicRootedOzoneClientAdapterImpl.java:182)
	at org.apache.hadoop.fs.ozone.RootedOzoneClientAdapterImpl.<init>(RootedOzoneClientAdapterImpl.java:51)
	at org.apache.hadoop.fs.ozone.RootedOzoneFileSystem.createAdapter(RootedOzoneFileSystem.java:93)
	at org.apache.hadoop.fs.ozone.BasicRootedOzoneFileSystem.initialize(BasicRootedOzoneFileSystem.java:150)
	at io.trino.hdfs.TrinoFileSystemCache.createFileSystem(TrinoFileSystemCache.java:155)
	at io.trino.hdfs.TrinoFileSystemCache$FileSystemHolder.createFileSystemOnce(TrinoFileSystemCache.java:293)
	at io.trino.hdfs.TrinoFileSystemCache.getInternal(TrinoFileSystemCache.java:135)
	at io.trino.hdfs.TrinoFileSystemCache.get(TrinoFileSystemCache.java:91)
	at org.apache.hadoop.fs.ForwardingFileSystemCache.get(ForwardingFileSystemCache.java:38)
	at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:479)
	at org.apache.hadoop.fs.Path.getFileSystem(Path.java:361)
	at io.trino.hdfs.HdfsEnvironment.lambda$getFileSystem$0(HdfsEnvironment.java:104)
	at io.trino.hdfs.authentication.NoHdfsAuthentication.doAs(NoHdfsAuthentication.java:25)
	at io.trino.hdfs.HdfsEnvironment.getFileSystem(HdfsEnvironment.java:103)
	at io.trino.hdfs.HdfsEnvironment.getFileSystem(HdfsEnvironment.java:97)
	at io.trino.filesystem.hdfs.HdfsInputFile.openFile(HdfsInputFile.java:113)
	at io.trino.filesystem.hdfs.HdfsInputFile.newStream(HdfsInputFile.java:69)
	at io.trino.filesystem.tracing.Tracing.withTracing(Tracing.java:47)
	at io.trino.filesystem.tracing.TracingInputFile.newStream(TracingInputFile.java:64)
	at io.trino.plugin.iceberg.fileio.ForwardingInputFile.newStream(ForwardingInputFile.java:52)
	at org.apache.iceberg.TableMetadataParser.read(TableMetadataParser.java:272)
	at io.trino.plugin.iceberg.catalog.AbstractIcebergTableOperations.lambda$refreshFromMetadataLocation$1(AbstractIcebergTableOperations.java:240)
	at dev.failsafe.Functions.lambda$toCtxSupplier$11(Functions.java:243)
	at dev.failsafe.Functions.lambda$get$0(Functions.java:46)
	at dev.failsafe.internal.RetryPolicyExecutor.lambda$apply$0(RetryPolicyExecutor.java:74)
	at dev.failsafe.SyncExecutionImpl.executeSync(SyncExecutionImpl.java:187)
	at dev.failsafe.FailsafeExecutor.call(FailsafeExecutor.java:376)
	at dev.failsafe.FailsafeExecutor.get(FailsafeExecutor.java:112)
	at io.trino.plugin.iceberg.catalog.AbstractIcebergTableOperations.refreshFromMetadataLocation(AbstractIcebergTableOperations.java:240)
	at io.trino.plugin.iceberg.catalog.AbstractIcebergTableOperations.refresh(AbstractIcebergTableOperations.java:134)
	at io.trino.plugin.iceberg.catalog.AbstractIcebergTableOperations.current(AbstractIcebergTableOperations.java:117)
	at io.trino.plugin.iceberg.catalog.hms.TrinoHiveCatalog.lambda$loadTable$11(TrinoHiveCatalog.java:374)
	at java.base/java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1708)
	at io.trino.plugin.iceberg.catalog.hms.TrinoHiveCatalog.loadTable(TrinoHiveCatalog.java:372)
	at io.trino.plugin.iceberg.IcebergMetadata.getTableHandle(IcebergMetadata.java:371)
	at io.trino.plugin.base.classloader.ClassLoaderSafeConnectorMetadata.getTableHandle(ClassLoaderSafeConnectorMetadata.java:1085)
	at io.trino.tracing.TracingConnectorMetadata.getTableHandle(TracingConnectorMetadata.java:143)
	at io.trino.metadata.MetadataManager.lambda$getTableHandle$5(MetadataManager.java:283)
	at java.base/java.util.Optional.flatMap(Optional.java:289)
	at io.trino.metadata.MetadataManager.getTableHandle(MetadataManager.java:277)
	at io.trino.metadata.MetadataManager.getRedirectionAwareTableHandle(MetadataManager.java:1572)
	at io.trino.metadata.MetadataManager.getRedirectionAwareTableHandle(MetadataManager.java:1564)
	at io.trino.tracing.TracingMetadata.getRedirectionAwareTableHandle(TracingMetadata.java:1272)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.getTableHandle(StatementAnalyzer.java:5411)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitTable(StatementAnalyzer.java:2183)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitTable(StatementAnalyzer.java:485)
	at io.trino.sql.tree.Table.accept(Table.java:60)
	at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:502)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.analyzeFrom(StatementAnalyzer.java:4470)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuerySpecification(StatementAnalyzer.java:2947)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuerySpecification(StatementAnalyzer.java:485)
	at io.trino.sql.tree.QuerySpecification.accept(QuerySpecification.java:155)
	at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:502)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:510)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuery(StatementAnalyzer.java:1467)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.visitQuery(StatementAnalyzer.java:485)
	at io.trino.sql.tree.Query.accept(Query.java:107)
	at io.trino.sql.tree.AstVisitor.process(AstVisitor.java:27)
	at io.trino.sql.analyzer.StatementAnalyzer$Visitor.process(StatementAnalyzer.java:502)
	at io.trino.sql.analyzer.StatementAnalyzer.analyze(StatementAnalyzer.java:464)
	at io.trino.sql.analyzer.Analyzer.analyze(Analyzer.java:97)
	at io.trino.sql.analyzer.Analyzer.analyze(Analyzer.java:86)
	at io.trino.execution.SqlQueryExecution.analyze(SqlQueryExecution.java:271)
	at io.trino.execution.SqlQueryExecution.<init>(SqlQueryExecution.java:206)
	at io.trino.execution.SqlQueryExecution$SqlQueryExecutionFactory.createQueryExecution(SqlQueryExecution.java:845)
	at io.trino.dispatcher.LocalDispatchQueryFactory.lambda$createDispatchQuery$0(LocalDispatchQueryFactory.java:153)
	at io.trino.$gen.Trino_419____20230623_144846_2.call(Unknown Source)
	at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131)
	at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:74)
	at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)

I have no idea how to integrate trino with apache ozone.

• Kidong.

[hchautrung]

Hi Im looking for an open source solution of open lakehouse using iceberg and Ozone(replace MinIO). Could you please to share the installation process and configuration or docker compose file.
Thank you very much

[cloudcheflabs]

Hi @hchautrung ,

I have shared ozone configuration with s3g and kerberos authentication to install ozone as standalone in the discussion of ozone apache/ozone#4996.
I am using ozone as s3 compatible object storage with rest catalog, trino, spark to handle iceberg tables for my data lakehouse platform.

If you want to handle iceberg tables on ozone as s3 compatible object storage,

• first, you need iceberg data catalogs like hive metastore, or rest catalog, or nessie.
• and then you can manage iceberg tables(create table, insert, update, delete, merge into, select, etc) with such as spark, flink, trino, dremio.

I am the creator of Chango which is a data lakehouse platform with iceberg , rest catalog, and ozone, trino gateway, spark, trino, storage security, etc. It is free to install / evaluate Chango, but you need license for commercial use.

• Kidong.

[hchautrung]

Hi @cloudcheflabs ,

Thanks for your help

[javaids33]

Were you ever able to solve this connection? I'm running into a similar issue with this.

[cloudcheflabs]

Hi @javaids33,

As mentioned in the previous answer, I have used s3g of ozone to access ozone via s3 api, that is, ozone will be used as s3 compatible object storage.
Please, see the apache/ozone#4996 in which ozone configuration with s3g and kerberos authentication to install ozone as standalone can be found.

[javaids33]

I'm able to connect via the s3g, but there are files already created with the ofs pattern. I'm trying to do an iceberg.system.register table and still running into this error.

Did you ever continue down the path of ofs connection through trino?

[cloudcheflabs]

@javaids33

I could not access ozone using ofs with trino iceberg catalog,
but I am able to access ozone using s3g with trino iceberg catalog.

Let's see my setup for trino iceberg catalog.

1. I am using s3g of ozone which will be accessed through s3 api.

2. I am using Iceberg REST Catalog. You may use another iceberg catalogs(https://iceberg.apache.org/concepts/catalog/) supported by iceberg.

3. I am using the following iceberg catalog configuration in trino.

connector.name=iceberg
iceberg.catalog.type=rest
iceberg.rest-catalog.uri=xxx
iceberg.rest-catalog.security=OAUTH2
iceberg.rest-catalog.oauth2.token=xxx
iceberg.rest-catalog.warehouse=xxx
hive.s3.aws-access-key=xxx
hive.s3.aws-secret-key=xxx
hive.s3.endpoint=xxx
hive.s3.region=xxx
hive.s3.path-style-access=true
hive.s3.ssl.enabled=true

[apotheque]

To make Ozone work with Trino over OFS, you need to properly shadow Protobuf in the ozonefs-hadoop3-client module of Ozone. Shadowing Protobuf to io.trino.hadoop.$internal, as Trino does, is sufficient to make it work with Hive. After building the JAR, simply place it in trino/plugin/hive/hdfs. For Iceberg, placing the properly shadowed JAR under trino/plugin/iceberg/hdfs might also work. See apache/ozone#7729 for details.