增加IgnoreNullWhenUpdate

Author: szlele

flink-connector-jdbc-gaussdb/src/main/java/org/apache/flink/connector/jdbc/gaussdb/database/dialect/GaussdbDialectConverter.java

@@ -20,7 +20,11 @@
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialectConverter;
+import org.apache.flink.connector.jdbc.gaussdb.table.GaussdbFieldObject;
+import org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatement;
 import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.LogicalTypeRoot;
@@ -109,4 +113,20 @@ private JdbcDeserializationConverter createGaussDBArrayConverter(ArrayType array
     public String converterName() {
         return "Gaussdb";
     }
+
+    @Override
+    public FieldNamedPreparedStatement toExternal(
+            RowData rowData, FieldNamedPreparedStatement statement) throws SQLException {
+        GenericRowData genericRowData = (GenericRowData) rowData;
+        for (int index = 0; index < rowData.getArity(); index++) {
+            Object field = genericRowData.getField(index);
+            int sourceIndex = index;
+            if (field != null && field instanceof GaussdbFieldObject) {
+                sourceIndex = ((GaussdbFieldObject) field).getIndex();
+                genericRowData.setField(index, ((GaussdbFieldObject) field).getField());
+            }
+            toExternalConverters[sourceIndex].serialize(genericRowData, index, statement);
+        }
+        return statement;
+    }
 }

flink-connector-jdbc-gaussdb/src/main/java/org/apache/flink/connector/jdbc/gaussdb/table/GaussdbExtendConnectorOptions.java

@@ -0,0 +1,18 @@
+package org.apache.flink.connector.jdbc.gaussdb.table;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions;
+
+/**
+ * Gaussdb connector options.
+ */
+public class GaussdbExtendConnectorOptions extends JdbcConnectorOptions {
+
+    public static final ConfigOption<Boolean> SINK_IGNORE_NULL_WHEN_UPDATE =
+            ConfigOptions.key("sink.ignore-null-when-update")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to ignore null values when updating records. If true, null values will not be updated to the database.");
+}

flink-connector-jdbc-gaussdb/src/main/java/org/apache/flink/connector/jdbc/gaussdb/table/GaussdbExtendOptions.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.jdbc.gaussdb.table;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/** JDBC sink batch options. */
+@PublicEvolving
+public class GaussdbExtendOptions implements Serializable {
+
+    private final boolean ignoreNullWhenUpdate;
+
+    private GaussdbExtendOptions(boolean ignoreNullWhenUpdate) {
+        this.ignoreNullWhenUpdate = ignoreNullWhenUpdate;
+    }
+
+    public boolean isIgnoreNullWhenUpdate() {
+        return ignoreNullWhenUpdate;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        GaussdbExtendOptions that = (GaussdbExtendOptions) o;
+        return ignoreNullWhenUpdate == that.ignoreNullWhenUpdate;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(ignoreNullWhenUpdate);
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static GaussdbExtendOptions defaults() {
+        return builder().build();
+    }
+
+    /** Builder for {@link GaussdbExtendOptions}. */
+    @PublicEvolving
+    public static final class Builder {
+        private boolean ignoreNullWhenUpdate = false;
+
+        public Builder withIgnoreNullWhenUpdate(boolean ignoreNullWhenUpdate) {
+            this.ignoreNullWhenUpdate = ignoreNullWhenUpdate;
+            return this;
+        }
+
+        public GaussdbExtendOptions build() {
+            return new GaussdbExtendOptions(ignoreNullWhenUpdate);
+        }
+    }
+}

flink-connector-jdbc-gaussdb/src/main/java/org/apache/flink/connector/jdbc/gaussdb/table/GaussdbFieldObject.java

@@ -0,0 +1,22 @@
+package org.apache.flink.connector.jdbc.gaussdb.table;
+
+/**
+ * GaussdbFieldObject.
+ */
+public class GaussdbFieldObject {
+    private final Object field;
+    private final int index;
+
+    public GaussdbFieldObject(int index, Object field) {
+        this.index = index;
+        this.field = field;
+    }
+
+    public Object getField() {
+        return field;
+    }
+
+    public int getIndex() {
+        return index;
+    }
+}