[fix] Fix compile

apache · Nov 11, 2024 · 4ed3bb2 · 4ed3bb2
1 parent 5b8db57
commit 4ed3bb2
Showing 1 changed file with 6 additions and 3 deletions.
diff --git a/.../src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java b/.../src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
@@ -100,18 +100,21 @@ protected void applySchemaChange(
         } else if (schemaChange instanceof SchemaChange.UpdateColumnType) {
             SchemaChange.UpdateColumnType updateColumnType =
                     (SchemaChange.UpdateColumnType) schemaChange;
+            Preconditions.checkState(
+                    updateColumnType.fieldNames().size() == 1,
+                    "Paimon CDC currently does not support nested type schema evolution.");
             TableSchema schema =
                     schemaManager
                             .latest()
                             .orElseThrow(
                                     () ->
                                             new RuntimeException(
                                                     "Table does not exist. This is unexpected."));
-            int idx = schema.fieldNames().indexOf(updateColumnType.fieldName());
+            int idx = schema.fieldNames().indexOf(updateColumnType.fieldNames().get(0));
             Preconditions.checkState(
                     idx >= 0,
                     "Field name "
-                            + updateColumnType.fieldName()
+                            + updateColumnType.fieldNames().get(0)
                             + " does not exist in table. This is unexpected.");
             DataType oldType = schema.fields().get(idx).type();
             DataType newType = updateColumnType.newDataType();
@@ -123,7 +126,7 @@ protected void applySchemaChange(
                     throw new UnsupportedOperationException(
                             String.format(
                                     "Cannot convert field %s from type %s to %s of Paimon table %s.",
-                                    updateColumnType.fieldName(),
+                                    updateColumnType.fieldNames().get(0),
                                     oldType,
                                     newType,
                                     identifier.getFullName()));