danielcweeks commented on code in PR #9641:
URL: https://github.com/apache/iceberg/pull/9641#discussion_r1501868634
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/IcebergWriter.java:
##########
@@ -77,8 +78,47 @@ public void write(SinkRecord record) {
}
private Record convertToRow(SinkRecord record) {
- // FIXME: update this when the record converter is added
- return null;
+ if (!config.evolveSchemaEnabled()) {
+ return recordConverter.convert(record.value());
+ }
+
+ SchemaUpdate.Consumer updates = new SchemaUpdate.Consumer();
+ Record row = recordConverter.convert(record.value(), updates);
+
+ if (!updates.empty()) {
+ // complete the current file
+ flush();
+ // apply the schema updates, this will refresh the table
+ SchemaUtils.applySchemaUpdates(table, updates);
+ // initialize a new writer with the new schema
+ initNewWriter();
+ // convert the row again, this time using the new table schema
+ row = recordConverter.convert(record.value(), null);
+ }
+
+ return row;
+ }
+
+ private Operation extractCdcOperation(Object recordValue, String cdcField) {
+ Object opValue = Utilities.extractFromRecordValue(recordValue, cdcField);
+
+ if (opValue == null) {
+ return Operation.INSERT;
+ }
+
+ String opStr = opValue.toString().trim().toUpperCase();
+ if (opStr.isEmpty()) {
+ return Operation.INSERT;
+ }
+
+ switch (opStr.charAt(0)) {
+ case 'U':
+ return Operation.UPDATE;
+ case 'D':
+ return Operation.DELETE;
+ default:
+ return Operation.INSERT;
Review Comment:
Should we assume insert here? I would expect that something has gone
terribly wrong if none of the checks found an op code. I could see this being
an obvious issue where someone has configured the op code for the wrote field
and is now just getting inserts. Seems safer to throw here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
