Extra java 17 code clean up

Signed-off-by: Marvin Froeder <marvin@datasqrl.com>

Author: velo

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlBaseJdbcRowConverter.java

@@ -48,7 +48,7 @@ public SqrlBaseJdbcRowConverter(RowType rowType) {
   protected JdbcSerializationConverter wrapIntoNullableExternalConverter(
       JdbcSerializationConverter jdbcSerializationConverter, LogicalType type) {
     if (type.getTypeRoot() == TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
-      int timestampWithTimezone = Types.TIMESTAMP_WITH_TIMEZONE;
+      var timestampWithTimezone = Types.TIMESTAMP_WITH_TIMEZONE;
       return (val, index, statement) -> {
         if (val == null || val.isNullAt(index) || LogicalTypeRoot.NULL.equals(type.getTypeRoot())) {
           statement.setNull(index, timestampWithTimezone);
@@ -62,15 +62,15 @@ protected JdbcSerializationConverter wrapIntoNullableExternalConverter(
 
   @Override
   public JdbcDeserializationConverter createInternalConverter(LogicalType type) {
-    LogicalTypeRoot root = type.getTypeRoot();
+    var root = type.getTypeRoot();
 
     if (root == LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
       return val ->
           val instanceof LocalDateTime ldt
               ? TimestampData.fromLocalDateTime(ldt)
               : TimestampData.fromTimestamp((Timestamp) val);
     } else if (root == LogicalTypeRoot.ARRAY) {
-      ArrayType arrayType = (ArrayType) type;
+      var arrayType = (ArrayType) type;
       return createArrayConverter(arrayType);
     } else if (root == LogicalTypeRoot.ROW) {
       return val -> val;
@@ -83,16 +83,15 @@ public JdbcDeserializationConverter createInternalConverter(LogicalType type) {
 
   @Override
   protected JdbcSerializationConverter createExternalConverter(LogicalType type) {
-    switch (type.getTypeRoot()) {
-      case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-        final int tsPrecision = ((LocalZonedTimestampType) type).getPrecision();
-        return (val, index, statement) ->
+    return switch (type.getTypeRoot()) {
+      case TIMESTAMP_WITH_LOCAL_TIME_ZONE -> {
+        final var tsPrecision = ((LocalZonedTimestampType) type).getPrecision();
+        yield (val, index, statement) ->
             statement.setTimestamp(index, val.getTimestamp(index, tsPrecision).toTimestamp());
-      case MULTISET:
-      case RAW:
-      default:
-        return super.createExternalConverter(type);
-    }
+      }
+      case MULTISET, RAW -> super.createExternalConverter(type);
+      default -> super.createExternalConverter(type);
+    };
   }
 
   @SneakyThrows

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlJdbcDynamicTableFactory.java

@@ -71,15 +71,14 @@ public class SqrlJdbcDynamicTableFactory implements DynamicTableSinkFactory {
 
   @Override
   public DynamicTableSink createDynamicTableSink(Context context) {
-    final FactoryUtil.TableFactoryHelper helper =
-        FactoryUtil.createTableFactoryHelper(this, context);
-    final ReadableConfig config = helper.getOptions();
+    final var helper = FactoryUtil.createTableFactoryHelper(this, context);
+    final var config = helper.getOptions();
 
     helper.validate();
     validateConfigOptions(config, context.getClassLoader());
     validateDataTypeWithJdbcDialect(
         context.getPhysicalRowDataType(), config.get(URL), context.getClassLoader());
-    InternalJdbcConnectionOptions jdbcOptions = getJdbcOptions(config, context.getClassLoader());
+    var jdbcOptions = getJdbcOptions(config, context.getClassLoader());
 
     return new JdbcDynamicTableSink(
         jdbcOptions,
@@ -91,15 +90,15 @@ public DynamicTableSink createDynamicTableSink(Context context) {
 
   private static void validateDataTypeWithJdbcDialect(
       DataType dataType, String url, ClassLoader classLoader) {
-    JdbcDialect dialect = loadDialect(url, classLoader);
+    var dialect = loadDialect(url, classLoader);
 
     dialect.validate((RowType) dataType.getLogicalType());
   }
 
   private InternalJdbcConnectionOptions getJdbcOptions(
       ReadableConfig readableConfig, ClassLoader classLoader) {
-    final String url = readableConfig.get(URL);
-    final InternalJdbcConnectionOptions.Builder builder =
+    final var url = readableConfig.get(URL);
+    final var builder =
         InternalJdbcConnectionOptions.builder()
             .setClassLoader(classLoader)
             .setDBUrl(url)
@@ -116,7 +115,7 @@ private InternalJdbcConnectionOptions getJdbcOptions(
   }
 
   private static JdbcDialect loadDialect(String url, ClassLoader classLoader) {
-    JdbcDialect dialect = JdbcDialectLoader.load(url, classLoader);
+    var dialect = JdbcDialectLoader.load(url, classLoader);
     // sqrl: standard postgres dialect with extended dialect
     if (dialect.dialectName().equalsIgnoreCase("PostgreSQL")) {
       return new SqrlPostgresDialect();
@@ -125,7 +124,7 @@ private static JdbcDialect loadDialect(String url, ClassLoader classLoader) {
   }
 
   private JdbcExecutionOptions getJdbcExecutionOptions(ReadableConfig config) {
-    final JdbcExecutionOptions.Builder builder = new JdbcExecutionOptions.Builder();
+    final var builder = new JdbcExecutionOptions.Builder();
     builder.withBatchSize(config.get(SINK_BUFFER_FLUSH_MAX_ROWS));
     builder.withBatchIntervalMs(config.get(SINK_BUFFER_FLUSH_INTERVAL).toMillis());
     builder.withMaxRetries(config.get(SINK_MAX_RETRIES));
@@ -135,7 +134,7 @@ private JdbcExecutionOptions getJdbcExecutionOptions(ReadableConfig config) {
   private JdbcDmlOptions getJdbcDmlOptions(
       InternalJdbcConnectionOptions jdbcOptions, DataType dataType, int[] primaryKeyIndexes) {
 
-    String[] keyFields =
+    var keyFields =
         Arrays.stream(primaryKeyIndexes)
             .mapToObj(i -> DataType.getFieldNames(dataType).get(i))
             .toArray(String[]::new);
@@ -209,7 +208,7 @@ public Set<ConfigOption<?>> forwardOptions() {
   }
 
   private void validateConfigOptions(ReadableConfig config, ClassLoader classLoader) {
-    String jdbcUrl = config.get(URL);
+    var jdbcUrl = config.get(URL);
     //        JdbcDialectLoader.load(jdbcUrl, classLoader);
 
     checkAllOrNone(config, new ConfigOption[] {USERNAME, PASSWORD});
@@ -263,14 +262,13 @@ private void validateConfigOptions(ReadableConfig config, ClassLoader classLoade
   }
 
   private void checkAllOrNone(ReadableConfig config, ConfigOption<?>[] configOptions) {
-    int presentCount = 0;
+    var presentCount = 0;
     for (ConfigOption configOption : configOptions) {
       if (config.getOptional(configOption).isPresent()) {
         presentCount++;
       }
     }
-    String[] propertyNames =
-        Arrays.stream(configOptions).map(ConfigOption::key).toArray(String[]::new);
+    var propertyNames = Arrays.stream(configOptions).map(ConfigOption::key).toArray(String[]::new);
     Preconditions.checkArgument(
         configOptions.length == presentCount || presentCount == 0,
         "Either all or none of the following options should be provided:\n"

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlPostgresDialect.java

@@ -68,9 +68,9 @@ public Optional<String> defaultDriverName() {
   @Override
   public Optional<String> getUpsertStatement(
       String tableName, String[] fieldNames, String[] uniqueKeyFields) {
-    String uniqueColumns =
+    var uniqueColumns =
         Arrays.stream(uniqueKeyFields).map(this::quoteIdentifier).collect(Collectors.joining(", "));
-    String updateClause =
+    var updateClause =
         Arrays.stream(fieldNames)
             .map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
             .collect(Collectors.joining(", "));

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/jdbc/SqrlPostgresRowConverter.java

@@ -99,8 +99,7 @@ public JdbcDeserializationConverter createArrayConverter(ArrayType arrayType) {
     // primitive byte arrays
     final Class<?> elementClass =
         LogicalTypeUtils.toInternalConversionClass(arrayType.getElementType());
-    final JdbcDeserializationConverter elementConverter =
-        createNullableInternalConverter(arrayType.getElementType());
+    final var elementConverter = createNullableInternalConverter(arrayType.getElementType());
     return val -> {
       // sqrl: check if scalar array
 
@@ -110,9 +109,8 @@ public JdbcDeserializationConverter createArrayConverter(ArrayType arrayType) {
       } else {
         in = (Object[]) val;
       }
-      final Object[] array =
-          (Object[]) java.lang.reflect.Array.newInstance(elementClass, in.length);
-      for (int i = 0; i < in.length; i++) {
+      final var array = (Object[]) java.lang.reflect.Array.newInstance(elementClass, in.length);
+      for (var i = 0; i < in.length; i++) {
         array[i] = elementConverter.deserialize(in[i]);
       }
       return new GenericArrayData(array);

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/type/FlinkArrayTypeUtil.java

@@ -29,34 +29,33 @@ public static LogicalType getBaseFlinkArrayType(LogicalType type) {
 
   public static boolean isScalarArray(LogicalType type) {
     if (type instanceof ArrayType arrayType) {
-      LogicalType elementType = arrayType.getElementType();
+      var elementType = arrayType.getElementType();
       return isScalar(elementType) || isScalarArray(elementType);
     }
     return false;
   }
 
   public static boolean isScalar(LogicalType type) {
-    switch (type.getTypeRoot()) {
-      case BOOLEAN:
-      case TINYINT:
-      case SMALLINT:
-      case INTEGER:
-      case BIGINT:
-      case FLOAT:
-      case DOUBLE:
-      case CHAR:
-      case VARCHAR:
-      case BINARY:
-      case VARBINARY:
-      case DATE:
-      case TIME_WITHOUT_TIME_ZONE:
-      case TIMESTAMP_WITH_TIME_ZONE:
-      case TIMESTAMP_WITHOUT_TIME_ZONE:
-      case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-      case DECIMAL:
-        return true;
-      default:
-        return false;
-    }
+    return switch (type.getTypeRoot()) {
+      case BOOLEAN,
+              TINYINT,
+              SMALLINT,
+              INTEGER,
+              BIGINT,
+              FLOAT,
+              DOUBLE,
+              CHAR,
+              VARCHAR,
+              BINARY,
+              VARBINARY,
+              DATE,
+              TIME_WITHOUT_TIME_ZONE,
+              TIMESTAMP_WITH_TIME_ZONE,
+              TIMESTAMP_WITHOUT_TIME_ZONE,
+              TIMESTAMP_WITH_LOCAL_TIME_ZONE,
+              DECIMAL ->
+          true;
+      default -> false;
+    };
   }
 }

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/type/PostgresArrayTypeConverter.java

@@ -22,56 +22,28 @@ public class PostgresArrayTypeConverter {
 
   /** Return the base array type for flink type */
   public static String getArrayScalarName(LogicalType type) {
-    switch (type.getTypeRoot()) {
-      case CHAR:
-      case VARCHAR:
-        return "text";
-      case BOOLEAN:
-        return "boolean";
-      case BINARY:
-      case VARBINARY:
-        return "bytea";
-      case DECIMAL:
-        return "decimal";
-      case TINYINT:
-        return "smallint";
-      case SMALLINT:
-        return "smallint";
-      case INTEGER:
-        return "integer";
-      case BIGINT:
-        return "bigint";
-      case FLOAT:
-        return "real"; // PostgreSQL uses REAL for float
-      case DOUBLE:
-        return "double";
-      case DATE:
-        return "date";
-      case TIME_WITHOUT_TIME_ZONE:
-        return "time without time zone";
-      case TIMESTAMP_WITHOUT_TIME_ZONE:
-        return "timestamp without time zone";
-      case TIMESTAMP_WITH_TIME_ZONE:
-      case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-        return "timestamptz";
-      case INTERVAL_YEAR_MONTH:
-        return "interval year to month";
-      case INTERVAL_DAY_TIME:
-        return "interval day to second";
-      case NULL:
-        return "void";
-      case ARRAY:
-        return getArrayScalarName(((ArrayType) type).getElementType());
-      case MULTISET:
-      case MAP:
-      case ROW:
-      case DISTINCT_TYPE:
-      case STRUCTURED_TYPE:
-      case RAW:
-      case SYMBOL:
-      case UNRESOLVED:
-      default:
-        throw new RuntimeException("Cannot convert type to array type");
-    }
+    return switch (type.getTypeRoot()) {
+      case CHAR, VARCHAR -> "text";
+      case BOOLEAN -> "boolean";
+      case BINARY, VARBINARY -> "bytea";
+      case DECIMAL -> "decimal";
+      case TINYINT -> "smallint";
+      case SMALLINT -> "smallint";
+      case INTEGER -> "integer";
+      case BIGINT -> "bigint";
+      case FLOAT -> "real"; // PostgreSQL uses REAL for float
+      case DOUBLE -> "double";
+      case DATE -> "date";
+      case TIME_WITHOUT_TIME_ZONE -> "time without time zone";
+      case TIMESTAMP_WITHOUT_TIME_ZONE -> "timestamp without time zone";
+      case TIMESTAMP_WITH_TIME_ZONE, TIMESTAMP_WITH_LOCAL_TIME_ZONE -> "timestamptz";
+      case INTERVAL_YEAR_MONTH -> "interval year to month";
+      case INTERVAL_DAY_TIME -> "interval day to second";
+      case NULL -> "void";
+      case ARRAY -> getArrayScalarName(((ArrayType) type).getElementType());
+      case MULTISET, MAP, ROW, DISTINCT_TYPE, STRUCTURED_TYPE, RAW, SYMBOL, UNRESOLVED ->
+          throw new RuntimeException("Cannot convert type to array type");
+      default -> throw new RuntimeException("Cannot convert type to array type");
+    };
   }
 }

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/type/PostgresJsonTypeSerializer.java

@@ -45,23 +45,23 @@ public String dialectTypeName() {
   public GenericDeserializationConverter<JdbcDeserializationConverter> getDeserializerConverter() {
     return () ->
         (val) -> {
-          FlinkJsonType t = (FlinkJsonType) val;
+          var t = (FlinkJsonType) val;
           return t.getJson();
         };
   }
 
   @Override
   public GenericSerializationConverter<JdbcSerializationConverter> getSerializerConverter(
       LogicalType type) {
-    FlinkJsonTypeSerializer typeSerializer = new FlinkJsonTypeSerializer();
+    var typeSerializer = new FlinkJsonTypeSerializer();
 
     return () ->
         (val, index, statement) -> {
           if (val != null && !val.isNullAt(index)) {
-            PGobject pgObject = new PGobject();
+            var pgObject = new PGobject();
             pgObject.setType("json");
             RawValueData<FlinkJsonType> object = val.getRawValue(index);
-            FlinkJsonType vec = object.toObject(typeSerializer);
+            var vec = object.toObject(typeSerializer);
             if (vec == null) {
               statement.setObject(index, null);
             } else {

connectors/postgresql-connector/src/main/java/com/datasqrl/connector/postgresql/type/PostgresRowTypeSerializer.java

@@ -20,9 +20,7 @@
 import org.apache.flink.connector.jdbc.converter.AbstractJdbcRowConverter.JdbcSerializationConverter;
 import org.apache.flink.formats.common.TimestampFormat;
 import org.apache.flink.formats.json.JsonFormatOptions.MapNullKeyMode;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.types.Row;
@@ -56,21 +54,21 @@ public GenericDeserializationConverter<JdbcDeserializationConverter> getDeserial
   @Override
   public GenericSerializationConverter<JdbcSerializationConverter> getSerializerConverter(
       LogicalType type) {
-    ObjectMapper mapper = new ObjectMapper();
+    var mapper = new ObjectMapper();
     return () ->
         (val, index, statement) -> {
           if (val != null && !val.isNullAt(index)) {
-            SqrlRowDataToJsonConverters rowDataToJsonConverter =
+            var rowDataToJsonConverter =
                 new SqrlRowDataToJsonConverters(TimestampFormat.SQL, MapNullKeyMode.DROP, "null");
 
-            ArrayType arrayType = (ArrayType) type;
-            ObjectNode objectNode = mapper.createObjectNode();
-            JsonNode convert =
+            var arrayType = (ArrayType) type;
+            var objectNode = mapper.createObjectNode();
+            var convert =
                 rowDataToJsonConverter
                     .createConverter(arrayType.getElementType())
                     .convert(mapper, objectNode, val);
 
-            PGobject pgObject = new PGobject();
+            var pgObject = new PGobject();
             pgObject.setType("json");
             pgObject.setValue(convert.toString());
             statement.setObject(index, pgObject);