Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: convert object value to linked hash map for Iceberg record #48726

Merged
merged 4 commits into from
Nov 27, 2024
Merged

fix: convert object value to linked hash map for Iceberg record #48726

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b300207
fix: convert object value to linked hash map for Iceberg record
jdpgrailsdev Nov 27, 2024
806f386
Merge branch 'master' into jonathan/iceberg-object-value-mapping
jdpgrailsdev Nov 27, 2024
5e13f57
Remove invalid import
jdpgrailsdev Nov 27, 2024
c492723
Avoid potential NPE
jdpgrailsdev Nov 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions ...t/src/main/kotlin/io/airbyte/cdk/load/data/iceberg/parquet/AirbyteValueToIcebergRecord.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,21 +27,25 @@ class AirbyteValueToIcebergRecord {
if (type.isStructType) {
type.asStructType().asSchema()
} else {
throw IllegalArgumentException("ObjectValue should be mapped to ObjectType")
throw IllegalArgumentException("ObjectValue should be mapped to StructType")
}
return recordSchema
.columns()
jdpgrailsdev marked this conversation as resolved.
Show resolved Hide resolved
.filter { column -> airbyteValue.values.containsKey(column.name()) }
.associate { column ->
column.name() to
convert(
airbyteValue.values[column.name()]!!,
column.type(),
)
}
val associate = recordSchema.columns().associate { it.name() to it.type() }
val record = GenericRecord.create(recordSchema)
airbyteValue.values.forEach { (name, value) ->
associate[name]?.let { field -> record.setField(name, convert(value, field)) }
}
return record
}
is ArrayValue -> {
val elementType =
if (type.isListType) {
type.asListType().elementType()
} else {
throw IllegalArgumentException("ArrayValue should be mapped to ArrayType")
throw IllegalArgumentException("ArrayValue should be mapped to ListType")
}

val array: MutableList<Any?> = mutableListOf()
Expand Down
78 changes: 69 additions & 9 deletions ...tures/kotlin/io/airbyte/cdk/load/data/icerberg/parquet/AirbyteValueToIcebergRecordTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,27 @@

package io.airbyte.cdk.load.data.icerberg.parquet

import io.airbyte.cdk.load.data.*
import io.airbyte.cdk.load.data.ArrayValue
import io.airbyte.cdk.load.data.BooleanValue
import io.airbyte.cdk.load.data.DateValue
import io.airbyte.cdk.load.data.IntegerValue
import io.airbyte.cdk.load.data.NullValue
import io.airbyte.cdk.load.data.NumberValue
import io.airbyte.cdk.load.data.ObjectValue
import io.airbyte.cdk.load.data.StringValue
import io.airbyte.cdk.load.data.TimeValue
import io.airbyte.cdk.load.data.TimestampValue
import io.airbyte.cdk.load.data.UnknownValue
import io.airbyte.cdk.load.data.iceberg.parquet.AirbyteValueToIcebergRecord
import io.airbyte.cdk.load.data.iceberg.parquet.toIcebergRecord
import io.airbyte.protocol.models.Jsons
import java.math.BigDecimal
import org.apache.iceberg.Schema
import org.apache.iceberg.data.GenericRecord
import org.apache.iceberg.types.Types
import org.apache.iceberg.types.Types.NestedField
import org.junit.jupiter.api.Assertions.*
import org.apache.iceberg.types.Types.StructType
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNull
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.assertThrows

Expand All @@ -31,9 +42,9 @@ class AirbyteValueToIcebergRecordTest {
val objectValue =
ObjectValue(linkedMapOf("id" to IntegerValue(42L), "name" to StringValue("John Doe")))

val result = converter.convert(objectValue, schema.asStruct()) as GenericRecord
assertEquals(42L, result.getField("id"))
assertEquals("John Doe", result.getField("name"))
val result = converter.convert(objectValue, schema.asStruct()) as LinkedHashMap<*, *>
assertEquals(42L, result["id"])
assertEquals("John Doe", result["name"])
}

@Test
Expand Down Expand Up @@ -128,22 +139,71 @@ class AirbyteValueToIcebergRecordTest {
val schema =
Schema(
NestedField.required(1, "id", Types.LongType.get()),
NestedField.optional(2, "name", Types.StringType.get())
NestedField.optional(2, "name", Types.StringType.get()),
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice. Thanks

NestedField.required(
3,
"meta",
Types.StructType.of(
NestedField.required(4, "sync_id", Types.IntegerType.get()),
NestedField.required(
5,
"changes",
StructType.of(
NestedField.required(6, "change", Types.StringType.get()),
NestedField.required(7, "reason", Types.StringType.get()),
)
)
)
)
)
val objectValue =
ObjectValue(linkedMapOf("id" to IntegerValue(123L), "name" to StringValue("John Doe")))
ObjectValue(
linkedMapOf(
"id" to IntegerValue(123L),
"name" to StringValue("John Doe"),
"meta" to
ObjectValue(
linkedMapOf(
"sync_id" to IntegerValue(123L),
"changes" to
ObjectValue(
linkedMapOf(
"change" to StringValue("insert"),
"reason" to StringValue("reason"),
)
)
)
)
)
)

val result = objectValue.toIcebergRecord(schema)
assertEquals(123L, result.getField("id"))
assertEquals("John Doe", result.getField("name"))
assertEquals(123L, (result.getField("meta") as Map<*, *>)["sync_id"] as Long)
assertEquals(
2,
((result.getField("meta") as Map<*, *>)["changes"] as LinkedHashMap<*, *>).size
)
assertEquals(
"insert",
((result.getField("meta") as Map<*, *>)["changes"] as LinkedHashMap<*, *>)["change"]
)
assertEquals(
"reason",
((result.getField("meta") as Map<*, *>)["changes"] as LinkedHashMap<*, *>)["reason"]
)
}

@Test
fun `toIcebergRecord ignores fields not in schema`() {
val schema = Schema(NestedField.required(1, "id", Types.LongType.get()))
val objectValue =
ObjectValue(
linkedMapOf("id" to IntegerValue(123L), "name" to StringValue("Should be ignored"))
linkedMapOf(
"id" to IntegerValue(123L),
"name" to StringValue("Should be ignored"),
)
)

val result = objectValue.toIcebergRecord(schema)
Expand Down
Loading