[Feature][Connector-V2][Google Sheets] Add Google Sheets option rules (#3364)

* [Feature][Connector-V2][Google Sheets] Add Google Sheets option rules

* [Feature][Connector-V2][Google Sheets] Add optional schema in optionRule

* [Feature][Connector-V2][Google Sheets] fix doc

Co-authored-by: Hisoka <fanjiaeminem@qq.com>

Author: TaoZex

docs/en/connector-v2/source/GoogleSheets.md

@@ -26,7 +26,8 @@ Used to read data from GoogleSheets.
 | service_account_key | string       | yes      | -             |
 | sheet_id            | string       | yes      | -             |
 | sheet_name          | string       | yes      | -             |
-| schema              | config       | yes      | -             |
+| range               | string       | yes      | -             |
+| schema              | config       | no       | -             |
 
 ### service_account_key [string]
 
@@ -40,6 +41,10 @@ sheet id in a Google Sheets URL
 
 the name of the sheet you want to import
 
+### range [string]
+
+the range of the sheet you want to import
+
 ### schema [config]
 
 #### fields [config]

seatunnel-connectors-v2/connector-google-sheets/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/config/SheetsConfig.java

@@ -17,10 +17,25 @@
 
 package org.apache.seatunnel.connectors.seatunnel.google.sheets.config;
 
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
 public class SheetsConfig {
 
-    public static final String SERVICE_ACCOUNT_KEY = "service_account_key";
-    public static final String SHEET_ID = "sheet_id";
-    public static final String SHEET_NAME = "sheet_name";
-    public static final String RANGE = "range";
+    public static final Option<String> SERVICE_ACCOUNT_KEY = Options.key("service_account_key")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Google Sheets login service account key");
+    public static final Option<String> SHEET_ID = Options.key("sheet_id")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Google Sheets sheet id");
+    public static final Option<String> SHEET_NAME = Options.key("sheet_name")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Google Sheets sheet name that you want to import");
+    public static final Option<String> RANGE = Options.key("range")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Google Sheets sheet range that you want to import");
 }

seatunnel-connectors-v2/connector-google-sheets/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/config/SheetsParameters.java

@@ -35,10 +35,10 @@ public class SheetsParameters implements Serializable {
     private String range;
 
     public SheetsParameters buildWithConfig(Config config) {
-        this.serviceAccountKey = config.getString(SheetsConfig.SERVICE_ACCOUNT_KEY).getBytes();
-        this.sheetId = config.getString(SheetsConfig.SHEET_ID);
-        this.sheetName = config.getString(SheetsConfig.SHEET_NAME);
-        this.range = config.getString(SheetsConfig.RANGE);
+        this.serviceAccountKey = config.getString(SheetsConfig.SERVICE_ACCOUNT_KEY.key()).getBytes();
+        this.sheetId = config.getString(SheetsConfig.SHEET_ID.key());
+        this.sheetName = config.getString(SheetsConfig.SHEET_NAME.key());
+        this.range = config.getString(SheetsConfig.RANGE.key());
         return this;
     }
 

seatunnel-connectors-v2/connector-google-sheets/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/source/SheetsSource.java

@@ -55,7 +55,7 @@ public String getPluginName() {
 
     @Override
     public void prepare(Config pluginConfig) throws PrepareFailException {
-        CheckResult checkResult = CheckConfigUtil.checkAllExists(pluginConfig, SheetsConfig.SERVICE_ACCOUNT_KEY, SheetsConfig.SHEET_ID, SheetsConfig.SHEET_NAME, SheetsConfig.RANGE, SeaTunnelSchema.SCHEMA.key());
+        CheckResult checkResult = CheckConfigUtil.checkAllExists(pluginConfig, SheetsConfig.SERVICE_ACCOUNT_KEY.key(), SheetsConfig.SHEET_ID.key(), SheetsConfig.SHEET_NAME.key(), SheetsConfig.RANGE.key(), SeaTunnelSchema.SCHEMA.key());
         if (!checkResult.isSuccess()) {
             throw new PrepareFailException(getPluginName(), PluginType.SOURCE, checkResult.getMsg());
         }

seatunnel-connectors-v2/connector-google-sheets/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/source/SheetsSourceFactory.java

@@ -0,0 +1,45 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.google.sheets.source;
+
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.api.table.factory.TableSourceFactory;
+import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
+import org.apache.seatunnel.connectors.seatunnel.google.sheets.config.SheetsConfig;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(Factory.class)
+public class SheetsSourceFactory implements TableSourceFactory {
+    @Override
+    public String factoryIdentifier() {
+        return "GoogleSheets";
+    }
+
+    @Override
+    public OptionRule optionRule() {
+        return OptionRule.builder()
+                .required(SheetsConfig.SERVICE_ACCOUNT_KEY)
+                .required(SheetsConfig.SHEET_ID)
+                .required(SheetsConfig.SHEET_NAME)
+                .required(SheetsConfig.RANGE)
+                .optional(SeaTunnelSchema.SCHEMA)
+                .build();
+    }
+}