Iceberg

Iceberg tables can be referenced via Deephaven Schemas using the Extended Storage feature. This provides immediate access for Deephaven to read existing Iceberg tables as Deephaven tables, including support for Iceberg schema evolution and nested Iceberg structures.

Configuration

The easiest way to configure Iceberg tables for Deephaven is to use the built-in inference provided by LoadTableOptions. Advanced users can customize the mapping with a Resolver, or with custom inference options.

A type or catalog-impl key is required for an Iceberg catalog. Additional parameters may be necessary depending on the type of catalog. Commonly configured properties may include warehouse and uri. See Iceberg Catalog properties and your specific catalog implementation for more details on configuration. An Iceberg table identifier is also required.

import io.deephaven.enterprise.iceberg.IcebergTableOptions
import io.deephaven.iceberg.util.BuildCatalogOptions
import io.deephaven.iceberg.util.LoadTableOptions

// The options to read an Iceberg Catalog.
catalogOptions = BuildCatalogOptions.builder()
    .name("MyCatalog")
    .putAllProperties([
        "type": "<type>",
        // ... additional properties here ...
    ])
    .build()

// The options to load an Iceberg Table from the catalog, uses default
// inference options.
tableOptions = LoadTableOptions.builder()
    .id("IcebergNamespace.IcebergTableName")
    .build()

// Load the Iceberg Catalog and Table and materialize the results into
// an explicit Resolver.
options = IcebergTableOptions.builder()
    .tableKey("DHNamespace", "DHTableName")
    .catalogOptions(catalogOptions)
    .tableOptions(tableOptions)
    .build()
    .materialize()

// Verify the resulting Table data looks correct
myTable = options.table()

If you have a working Spark configuration, that can typically be translated into the necessary Catalog properties by removing the Spark prefix.

For example, the following Spark properties:

spark.sql.catalog.rest_prod = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.rest_prod.type = rest
spark.sql.catalog.rest_prod.uri = http://localhost:8080

Translate into the given BuildCatalogOptions:

catalogOptions = BuildCatalogOptions.builder()
    .name("rest_prod")
    .putAllProperties([
        "type": "rest",
        "uri": "http://localhost:8080"
    ])
    .build()

See BuildCatalogOptions and LoadTableOptions for more details on these structures.

Deployment

An iris-schemamanagers user is required to deploy the Schema.

import com.illumon.iris.db.schema.SchemaServiceFactory

schemaService = SchemaServiceFactory.getDefault()

// Admins are encouraged to explicitly manage the deployment logic of their Schema with
// the SchemaService.
mySchema = options.schema().get()
schemaService.addSchema(mySchema)

// Alternatively, admins may deploy the schema directly using the IcebergTableOptions.
// The following will create the schema namespace if it doesn't exist, create the schema if it doesn't exist, or update it if it does.
// options.deploy(schemaService)

// Verify that the table can be fetched
myTableViaDb = db.historicalTable("DHNamespace", "DHTableName")

Serialization format

Caution

The creation and deployment of a Deephaven Iceberg schema is typically performed programmatically, as shown in the previous sections. Exercise caution when manually creating or editing a schema.

An Iceberg table is referenced in a Deephaven table's schema using an ExtendedStorage element with the attribute type set to iceberg.

<Table namespace="DHNamespace" name="DHTableName" namespaceSet="System" storageType="Extended">
  <!-- Column elements omitted for brevity -->
  <ExtendedStorage type="iceberg">
    <Catalog><!-- see `Catalog` section below --></Catalog>
    <Table><!-- see `Table` section below --></Table>
  </ExtendedStorage>
</Table>

`Catalog` element

The Catalog element is a serialization of the core BuildCatalogOptions. It is composed of a Name, Properties, and optional HadoopConfig element. The Properties element is a map of string keys to string values. The optional HadoopConfig element is optional and is an additional map for Hadoop catalogs. For example:

<Catalog>
  <Name>MyCatalog</Name>
  <Properties injection="enabled">
    <Entry key="type" value="<type>" />
    <!--
    <Entry key="warehouse" value="..." />
    -->
  </Properties>
  <!--
  <HadoopConfig>
    <Entry key="..." value="...">
  </HadoopConfig>
  -->
</Catalog>

The injection attribute on the Properties element controls whether Deephaven may automatically add properties that work around known upstream issues and/or supply defaults needed for Deephaven's Iceberg usage. The valid values are enabled and disabled. It is recommended to set this to enabled.

`Table` element

The Table element is a serialization of the core LoadTableOptions. It is composed of a TableIdentifier, Resolver, and NameMapping element.

<Table>
  <TableIdentifier>IcebergNamespace.IcebergTableName</TableIdentifier>
  <Resolver><!-- see `Resolver` section below --></Resolver>
  <NameMapping><!-- see `NameMapping` section below --></NameMapping>
</Table>

The Resolver element contains a ColumnInstructions, Schema, and optional PartitionSpec element. The ColumnInstructions element contains the mapping from Deephaven column names to Iceberg fieldId, Iceberg partitionFieldId, or type unmapped. The Schema element contains the Iceberg Schema JSON. The optional PartitionSpec element contains the Iceberg Partition Spec JSON.

<Resolver type="direct">
  <ColumnInstructions>
    <Column name="Foo" partitionFieldId="1000" />
    <Column name="Bar" fieldId="2" />
    <Column name="Baz" type="unmapped" />
  </ColumnInstructions>
  <Schema type="json"><![CDATA[{
  "type" : "struct",
  "schema-id" : 0,
  "fields" : [ {
    "id" : 1,
    "name" : "foo",
    "required" : false,
    "type" : "int"
  }, {
    "id" : 2,
    "name" : "bar",
    "required" : false,
    "type" : "long"
  } ]
}]]></Schema>
  <PartitionSpec type="json"><![CDATA[{
  "spec-id" : 0,
  "fields" : [ {
    "name" : "foo",
    "transform" : "identity",
    "source-id" : 1,
    "field-id" : 1000
  } ]
}]]></PartitionSpec>
</Resolver>

The NameMapping element provides fallback field ids to be used when a data file does not contain field id information. It has three different types, specified via the type attribute.

The table type means to read the Name Mapping from the Iceberg Table property schema.name-mapping.default (see https://iceberg.apache.org/spec/#column-projection).

<NameMapping type="table" />

The empty type means to not use name mapping.

<NameMapping type="empty" />

The json type uses Iceberg Name Mapping JSON.

<NameMapping type="json"><![CDATA[[ {
  "field-id" : 1,
  "names" : [ "Foo" ]
}, {
  "field-id" : 2,
  "names" : [ "Bar" ]
} ]]]></NameMapping>

Full example

Let's assume that we have an existing Iceberg Glue Catalog that contains a table mycatalog.cities with the Iceberg Schema:

{
  "type": "struct",
  "schema-id": 1,
  "fields": [
    {
      "id": 1,
      "name": "city",
      "required": false,
      "type": "string"
    },
    {
      "id": 2,
      "name": "latitude",
      "required": false,
      "type": "double"
    },
    {
      "id": 3,
      "name": "longitude",
      "required": false,
      "type": "double"
    }
  ]
}

To create a new Deephaven Schema with namespace DhExample and table name Cities that references this Iceberg Table, we would execute the following once:

import io.deephaven.enterprise.iceberg.IcebergTableOptions
import io.deephaven.iceberg.util.BuildCatalogOptions
import io.deephaven.iceberg.util.LoadTableOptions
import com.illumon.iris.db.schema.SchemaServiceFactory

catalogOptions = BuildCatalogOptions.builder()
    .name("GlueCatalog")
    .putAllProperties(["type": "glue"])
    .build()

tableOptions = LoadTableOptions.builder()
    .id("mycatalog.cities")
    .build()

options = IcebergTableOptions.builder()
    .tableKey("DhExample", "Cities")
    .catalogOptions(catalogOptions)
    .tableOptions(tableOptions)
    .build()
    .materialize()

SchemaServiceFactory.getDefault().addSchema(options.schema().get())

This would result in the following Deephaven Schema:

<Table namespace="DhExample" name="Cities" namespaceSet="System" storageType="Extended">
  <Column name="city" dataType="String" columnType="Normal" />
  <Column name="latitude" dataType="double" columnType="Normal" />
  <Column name="longitude" dataType="double" columnType="Normal" />
  <ExtendedStorage type="iceberg">
    <Catalog>
      <Name>GlueCatalog</Name>
      <Properties injection="enabled">
        <Entry key="glue" />
      </Properties>
    </Catalog>
    <Table>
      <TableIdentifier>mycatalog.cities</TableIdentifier>
      <Resolver type="direct">
        <ColumnInstructions>
          <Column name="city" fieldId="1" />
          <Column name="latitude" fieldId="2" />
          <Column name="longitude" fieldId="3" />
        </ColumnInstructions>
        <Schema type="json"><![CDATA[{
  "type" : "struct",
  "schema-id" : 1,
  "fields" : [ {
    "id" : 1,
    "name" : "city",
    "required" : false,
    "type" : "string"
  }, {
    "id" : 2,
    "name" : "latitude",
    "required" : false,
    "type" : "double"
  }, {
    "id" : 3,
    "name" : "longitude",
    "required" : false,
    "type" : "double"
  } ]
}]]></Schema>
      </Resolver>
      <NameMapping type="empty" />
    </Table>
  </ExtendedStorage>
</Table>