Skip to content

Commit 48ede61

Browse files
liunaijieliunaijie
authored
[improve] google sheets options (#8922)
1 parent 45345f2 commit 48ede61

File tree

5 files changed

+53
-65
lines changed

5 files changed

+53
-65
lines changed

Diff for: ‎seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java

-1
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,6 @@ private Set<String> buildWhiteList() {
214214
whiteList.add("Neo4jSourceOptions");
215215
whiteList.add("HttpSourceOptions");
216216
whiteList.add("QdrantSourceOptions");
217-
whiteList.add("SheetsSourceOptions");
218217
whiteList.add("SocketSourceOptions");
219218
whiteList.add("OpenMldbSourceOptions");
220219
whiteList.add("Web3jSourceOptions");

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

+6-7
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.google.sheets.config;
1919

20-
import org.apache.seatunnel.shade.com.typesafe.config.Config;
20+
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
2121

2222
import lombok.Data;
2323

@@ -34,12 +34,11 @@ public class SheetsParameters implements Serializable {
3434

3535
private String range;
3636

37-
public SheetsParameters buildWithConfig(Config config) {
38-
this.serviceAccountKey =
39-
config.getString(SheetsConfig.SERVICE_ACCOUNT_KEY.key()).getBytes();
40-
this.sheetId = config.getString(SheetsConfig.SHEET_ID.key());
41-
this.sheetName = config.getString(SheetsConfig.SHEET_NAME.key());
42-
this.range = config.getString(SheetsConfig.RANGE.key());
37+
public SheetsParameters buildWithConfig(ReadonlyConfig config) {
38+
this.serviceAccountKey = config.get(SheetsSourceOptions.SERVICE_ACCOUNT_KEY).getBytes();
39+
this.sheetId = config.get(SheetsSourceOptions.SHEET_ID);
40+
this.sheetName = config.get(SheetsSourceOptions.SHEET_NAME);
41+
this.range = config.get(SheetsSourceOptions.RANGE);
4342
return this;
4443
}
4544
}

Diff for: ‎seatunnel-connectors-v2/connector-google-sheets/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/config/SheetsConfig.java renamed to ‎seatunnel-connectors-v2/connector-google-sheets/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/config/SheetsSourceOptions.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
import org.apache.seatunnel.api.configuration.Option;
2121
import org.apache.seatunnel.api.configuration.Options;
2222

23-
public class SheetsConfig {
23+
public class SheetsSourceOptions {
2424

2525
public static final Option<String> SERVICE_ACCOUNT_KEY =
2626
Options.key("service_account_key")

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

+17-51
Original file line numberDiff line numberDiff line change
@@ -17,73 +17,36 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.google.sheets.source;
1919

20-
import org.apache.seatunnel.shade.com.typesafe.config.Config;
21-
22-
import org.apache.seatunnel.api.common.PrepareFailException;
23-
import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
24-
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
2520
import org.apache.seatunnel.api.serialization.DeserializationSchema;
2621
import org.apache.seatunnel.api.source.Boundedness;
27-
import org.apache.seatunnel.api.source.SeaTunnelSource;
2822
import org.apache.seatunnel.api.table.catalog.CatalogTable;
29-
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
30-
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
3123
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
32-
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
33-
import org.apache.seatunnel.common.config.CheckConfigUtil;
34-
import org.apache.seatunnel.common.config.CheckResult;
35-
import org.apache.seatunnel.common.constants.PluginType;
3624
import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
3725
import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitSource;
3826
import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
39-
import org.apache.seatunnel.connectors.seatunnel.google.sheets.config.SheetsConfig;
4027
import org.apache.seatunnel.connectors.seatunnel.google.sheets.config.SheetsParameters;
41-
import org.apache.seatunnel.connectors.seatunnel.google.sheets.exception.GoogleSheetsConnectorException;
4228
import org.apache.seatunnel.format.json.JsonDeserializationSchema;
4329

44-
import com.google.auto.service.AutoService;
30+
import java.util.Collections;
31+
import java.util.List;
4532

46-
@AutoService(SeaTunnelSource.class)
4733
public class SheetsSource extends AbstractSingleSplitSource<SeaTunnelRow> {
4834

49-
private SeaTunnelRowType seaTunnelRowType;
50-
private CatalogTable catalogTable;
35+
private final CatalogTable catalogTable;
5136

52-
private SheetsParameters sheetsParameters;
37+
private final SheetsParameters sheetsParameters;
5338

54-
private DeserializationSchema<SeaTunnelRow> deserializationSchema;
39+
private final DeserializationSchema<SeaTunnelRow> deserializationSchema;
5540

56-
@Override
57-
public String getPluginName() {
58-
return "GoogleSheets";
41+
public SheetsSource(CatalogTable catalogTable, SheetsParameters sheetsParameters) {
42+
this.catalogTable = catalogTable;
43+
this.sheetsParameters = sheetsParameters;
44+
this.deserializationSchema = new JsonDeserializationSchema(catalogTable, false, false);
5945
}
6046

6147
@Override
62-
public void prepare(Config pluginConfig) throws PrepareFailException {
63-
CheckResult checkResult =
64-
CheckConfigUtil.checkAllExists(
65-
pluginConfig,
66-
SheetsConfig.SERVICE_ACCOUNT_KEY.key(),
67-
SheetsConfig.SHEET_ID.key(),
68-
SheetsConfig.SHEET_NAME.key(),
69-
SheetsConfig.RANGE.key(),
70-
ConnectorCommonOptions.SCHEMA.key());
71-
if (!checkResult.isSuccess()) {
72-
throw new GoogleSheetsConnectorException(
73-
SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
74-
String.format(
75-
"PluginName: %s, PluginType: %s, Message: %s",
76-
getPluginName(), PluginType.SOURCE, checkResult.getMsg()));
77-
}
78-
this.sheetsParameters = new SheetsParameters().buildWithConfig(pluginConfig);
79-
if (pluginConfig.hasPath(ConnectorCommonOptions.SCHEMA.key())) {
80-
this.catalogTable = CatalogTableUtil.buildWithConfig(pluginConfig);
81-
} else {
82-
this.catalogTable = CatalogTableUtil.buildSimpleTextTable();
83-
}
84-
85-
this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
86-
this.deserializationSchema = new JsonDeserializationSchema(catalogTable, false, false);
48+
public String getPluginName() {
49+
return "GoogleSheets";
8750
}
8851

8952
@Override
@@ -92,14 +55,17 @@ public Boundedness getBoundedness() {
9255
}
9356

9457
@Override
95-
public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
96-
return seaTunnelRowType;
58+
public List<CatalogTable> getProducedCatalogTables() {
59+
return Collections.singletonList(catalogTable);
9760
}
9861

9962
@Override
10063
public AbstractSingleSplitReader<SeaTunnelRow> createReader(
10164
SingleSplitReaderContext readerContext) throws Exception {
10265
return new SheetsSourceReader(
103-
sheetsParameters, readerContext, deserializationSchema, this.seaTunnelRowType);
66+
sheetsParameters,
67+
readerContext,
68+
deserializationSchema,
69+
catalogTable.getSeaTunnelRowType());
10470
}
10571
}

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

+29-5
Original file line numberDiff line numberDiff line change
@@ -20,12 +20,20 @@
2020
import org.apache.seatunnel.api.configuration.util.OptionRule;
2121
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
2222
import org.apache.seatunnel.api.source.SeaTunnelSource;
23+
import org.apache.seatunnel.api.source.SourceSplit;
24+
import org.apache.seatunnel.api.table.catalog.CatalogTable;
25+
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
26+
import org.apache.seatunnel.api.table.connector.TableSource;
2327
import org.apache.seatunnel.api.table.factory.Factory;
2428
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
25-
import org.apache.seatunnel.connectors.seatunnel.google.sheets.config.SheetsConfig;
29+
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
30+
import org.apache.seatunnel.connectors.seatunnel.google.sheets.config.SheetsParameters;
31+
import org.apache.seatunnel.connectors.seatunnel.google.sheets.config.SheetsSourceOptions;
2632

2733
import com.google.auto.service.AutoService;
2834

35+
import java.io.Serializable;
36+
2937
@AutoService(Factory.class)
3038
public class SheetsSourceFactory implements TableSourceFactory {
3139
@Override
@@ -36,14 +44,30 @@ public String factoryIdentifier() {
3644
@Override
3745
public OptionRule optionRule() {
3846
return OptionRule.builder()
39-
.required(SheetsConfig.SERVICE_ACCOUNT_KEY)
40-
.required(SheetsConfig.SHEET_ID)
41-
.required(SheetsConfig.SHEET_NAME)
42-
.required(SheetsConfig.RANGE)
47+
.required(SheetsSourceOptions.SERVICE_ACCOUNT_KEY)
48+
.required(SheetsSourceOptions.SHEET_ID)
49+
.required(SheetsSourceOptions.SHEET_NAME)
50+
.required(SheetsSourceOptions.RANGE)
4351
.optional(ConnectorCommonOptions.SCHEMA)
4452
.build();
4553
}
4654

55+
@Override
56+
public <T, SplitT extends SourceSplit, StateT extends Serializable>
57+
TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
58+
CatalogTable catalogTable;
59+
if (context.getOptions().getOptional(ConnectorCommonOptions.SCHEMA).isPresent()) {
60+
catalogTable = CatalogTableUtil.buildWithConfig(context.getOptions());
61+
} else {
62+
catalogTable = CatalogTableUtil.buildSimpleTextTable();
63+
}
64+
return () ->
65+
(SeaTunnelSource<T, SplitT, StateT>)
66+
new SheetsSource(
67+
catalogTable,
68+
new SheetsParameters().buildWithConfig(context.getOptions()));
69+
}
70+
4771
@Override
4872
public Class<? extends SeaTunnelSource> getSourceClass() {
4973
return SheetsSource.class;

0 commit comments

Comments
 (0)