SNOW-1737840: Adapt record mapping in RecordService #969
Conversation
| if (includeAllMetadata) { | ||
| streamingIngestRow.put(TABLE_COLUMN_METADATA, mapper.writeValueAsString(row.getMetadata())); | ||
| } |
There was a problem hiding this comment.
not sure if there was some trick here but i moved it from the forEach loop above to here. I don't see any sense in parsing and putting it all over again for every jsonNode. This should improve overall processing speed as we do not serialise it for every json node we have
There was a problem hiding this comment.
I think I even saw some tech debt ticket regarding multiple serialization. It's great that you've optimized it.
| import org.junit.jupiter.params.provider.Arguments; | ||
| import org.junit.jupiter.params.provider.MethodSource; | ||
|
|
||
| class IcebergTableStreamingRecordMapperTest { |
There was a problem hiding this comment.
please think of more test cases if possible with example :)
There was a problem hiding this comment.
Looks exhaustive to me, let's wait for the bugs to happen :)
|
|
||
| @Override | ||
| public Map<String, Object> processSnowflakeRecord( | ||
| SnowflakeTableRow row, boolean schematizationEnabled, boolean includeAllMetadata) |
There was a problem hiding this comment.
nit: maybe just boolean includeMetadata?
| String key = fields.next(); | ||
| JsonNode valueNode = headersNode.get(key); | ||
| String value; | ||
| if (valueNode.isTextual()) { |
There was a problem hiding this comment.
nit: extract String getTextualValue(JsonNode node) and reuse in SnowflakeTableStreamingRecordMapper?
| this.clock = clock; | ||
| this.enableSchematization = enableSchematization; |
There was a problem hiding this comment.
I wonder if RecordService needs to be aware of enableSchematization value after creating StreamingRecordMapper abstraction. In fact we could pass this flag to a mapper in a factory instead of processSnowflakeRecord argument.
| @@ -67,8 +123,7 @@ public static List<RecordWithMetadata<PrimitiveJsonRecord>> fromSchematizedResul | |||
| resultSet.getLong("ID_INT8"), | |||
| resultSet.getLong("ID_INT16"), | |||
| resultSet.getLong("ID_INT32"), | |||
| // FIXME: there is currently some bug in Iceberg when storing int64 values | |||
| // resultSet.getLong("ID_INT64"), | |||
| resultSet.getLong("ID_INT64"), | |||
There was a problem hiding this comment.
Thx, please close https://snowflakecomputing.atlassian.net/browse/SNOW-1754474 if it works.
sfc-gh-wtrefon
deleted the
wtrefon/SNOW-1737840-record-service-iceberg-mapping
branch
Overview
SNOW-1737840
Pre-review checklist
snowflake.ingestion.method.Yes- Added end to end and Unit Tests.No- Suggest why it is not param protected