Hi folks, is there an easy to convert

DataStream<rowdata>

generated from iceberg source to

DataStream<row>

? I am using

RowRowConverter

Nothing simpler than

RowRowConverter

in my opinion 🙂

but it doesn't return exact schema, I get raw type

Then you’re doing something wrong

I use

sourceTable.getResolvedSchema().toSourceRowDataType()

where

sourcetable

is

Table sourceTable = tableEnv.fromDataStream(icebergSource);

Ah, I remember now. In Scala, you also must provide

RowTypeInfo

as an implicit if you use

RowRowConverter

in something like a

map

or `flatMap`:

implicit val rowTypeInfo: RowTypeInfo = DataTypeUtils.createRowTypeInfoFromSchema(tableSchema)

Where

DataTypeUtils.createRowTypeInfoFromSchema

is just my utility to create

RowTypeInfo

from a schema. Not sure what Java alternative to this is.

how do you convert datatype to RowTypeInfo? Is there any flink converter class I can use?

This is my logic in Flink 1.15, but it does use some deprecated classes

def createRowTypeInfoFromSchema(schema: Schema): RowTypeInfo = {
    val tableColumns = schema.getColumns.asScala.toList.map(rawColumn => {
      rawColumn.asInstanceOf[UnresolvedPhysicalColumn]
    })

    new RowTypeInfo(
      tableColumns.map(_.getDataType.asInstanceOf[DataType]).map(TypeInfoDataTypeConverter.fromDataTypeToTypeInfo).toArray,
      tableColumns.map(_.getName).toArray
    )
  }

thank you! let me try that