diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java index 920e44b24b31..f92095963255 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java @@ -48,7 +48,14 @@ public RowDataWrapper(RowType rowType, Types.StructType struct) { for (int i = 0; i < size; i++) { types[i] = rowType.getTypeAt(i); - getters[i] = buildGetter(types[i], struct.fields().get(i).type()); + PositionalGetter getter = buildGetter(types[i], struct.fields().get(i).type()); + if (getter == null) { + // Pre-build the Flink field getter once instead of recreating it on every access. + RowData.FieldGetter fieldGetter = FlinkRowData.createFieldGetter(types[i], i); + getter = (row, pos) -> fieldGetter.getFieldOrNull(row); + } + + getters[i] = getter; } } @@ -66,12 +73,9 @@ public int size() { public T get(int pos, Class javaClass) { if (rowData.isNullAt(pos)) { return null; - } else if (getters[pos] != null) { - return javaClass.cast(getters[pos].get(rowData, pos)); } - Object value = FlinkRowData.createFieldGetter(types[pos], pos).getFieldOrNull(rowData); - return javaClass.cast(value); + return javaClass.cast(getters[pos].get(rowData, pos)); } @Override diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java index 920e44b24b31..f92095963255 100644 --- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java +++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java @@ -48,7 +48,14 @@ public RowDataWrapper(RowType rowType, Types.StructType struct) { for (int i = 0; i < size; i++) { types[i] = rowType.getTypeAt(i); - getters[i] = buildGetter(types[i], struct.fields().get(i).type()); + PositionalGetter getter = buildGetter(types[i], struct.fields().get(i).type()); + if (getter == null) { + // Pre-build the Flink field getter once instead of recreating it on every access. + RowData.FieldGetter fieldGetter = FlinkRowData.createFieldGetter(types[i], i); + getter = (row, pos) -> fieldGetter.getFieldOrNull(row); + } + + getters[i] = getter; } } @@ -66,12 +73,9 @@ public int size() { public T get(int pos, Class javaClass) { if (rowData.isNullAt(pos)) { return null; - } else if (getters[pos] != null) { - return javaClass.cast(getters[pos].get(rowData, pos)); } - Object value = FlinkRowData.createFieldGetter(types[pos], pos).getFieldOrNull(rowData); - return javaClass.cast(value); + return javaClass.cast(getters[pos].get(rowData, pos)); } @Override diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java index 920e44b24b31..f92095963255 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/RowDataWrapper.java @@ -48,7 +48,14 @@ public RowDataWrapper(RowType rowType, Types.StructType struct) { for (int i = 0; i < size; i++) { types[i] = rowType.getTypeAt(i); - getters[i] = buildGetter(types[i], struct.fields().get(i).type()); + PositionalGetter getter = buildGetter(types[i], struct.fields().get(i).type()); + if (getter == null) { + // Pre-build the Flink field getter once instead of recreating it on every access. + RowData.FieldGetter fieldGetter = FlinkRowData.createFieldGetter(types[i], i); + getter = (row, pos) -> fieldGetter.getFieldOrNull(row); + } + + getters[i] = getter; } } @@ -66,12 +73,9 @@ public int size() { public T get(int pos, Class javaClass) { if (rowData.isNullAt(pos)) { return null; - } else if (getters[pos] != null) { - return javaClass.cast(getters[pos].get(rowData, pos)); } - Object value = FlinkRowData.createFieldGetter(types[pos], pos).getFieldOrNull(rowData); - return javaClass.cast(value); + return javaClass.cast(getters[pos].get(rowData, pos)); } @Override