Skip to content

[improve] Adjust CosFile connector instantiation #9121

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
misi1987107 wants to merge 8 commits into from
Closed

[improve] Adjust CosFile connector instantiation #9121

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8e86af1
[improve] Adjust the instantiation method of CosFile connector
Apr 8, 2025
776e8b7
[improve] code format
Apr 8, 2025
b99661f
[improve] implements getProducedCatalogTables() method
Apr 8, 2025
a9fc8ca
[improve] modify deprecated method
Apr 9, 2025
c0baac8
Revert "[improve] modify deprecated method"
Apr 9, 2025
98fac43
[bugfix] remove CosFileSource prepare()
Apr 10, 2025
7b8edfc
[improve] remove CosFileSink prepare()
Apr 10, 2025
37f68aa
[improve] skip deprecated method check
Apr 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 0 additions & 6 deletions ...-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/BaseFileSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,12 +75,6 @@ public void setJobContext(JobContext jobContext) {
preCheckConfig();
}

@Override
public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
this.seaTunnelRowType = seaTunnelRowType;
this.fileSinkConfig = new FileSinkConfig(pluginConfig, seaTunnelRowType);
}
Copy link
Member

@Hisoka-X Hisoka-X Apr 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why delete this? We still have some sink should invoke it like OssFileSink, HdfsFileSink etc.

Copy link
Contributor Author

@misi1987107 misi1987107 Apr 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry,I neet to make adjustments according to purpose2 ,move the connector creation logic from Connector to ConnectorFactory.I don't know how to adjust it
(#8576 (comment))


@Override
public SinkWriter<SeaTunnelRow, FileCommitInfo, FileSinkState> restoreWriter(
SinkWriter.Context context, List<FileSinkState> states) {
Expand Down
12 changes: 6 additions & 6 deletions ...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import org.apache.seatunnel.api.source.SourceSplitEnumerator;
import org.apache.seatunnel.api.source.SupportColumnProjection;
import org.apache.seatunnel.api.source.SupportParallelism;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
Expand All @@ -48,11 +48,6 @@ public Boundedness getBoundedness() {
return Boundedness.BOUNDED;
}

@Override
public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
return rowType;
}

@Override
public SourceReader<SeaTunnelRow, FileSourceSplit> createReader(
SourceReader.Context readerContext) {
Expand All @@ -72,4 +67,9 @@ public SourceSplitEnumerator<FileSourceSplit, FileSourceState> restoreEnumerator
throws Exception {
return new FileSourceSplitEnumerator(enumeratorContext, filePaths, checkpointState);
}

@Override
public List<CatalogTable> getProducedCatalogTables() {
return SeaTunnelSource.super.getProducedCatalogTables();
}
Copy link
Member

@Hisoka-X Hisoka-X Apr 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should return catalog table in here to replace getProducedType method.

}
13 changes: 8 additions & 5 deletions ...os/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/sink/CosFileSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@

import org.apache.seatunnel.api.common.PrepareFailException;
import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.sink.SeaTunnelSink;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.common.config.CheckConfigUtil;
import org.apache.seatunnel.common.config.CheckResult;
Expand All @@ -33,12 +32,16 @@
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
import org.apache.seatunnel.connectors.seatunnel.file.sink.BaseFileSink;

import com.google.auto.service.AutoService;

import java.util.Optional;

@AutoService(SeaTunnelSink.class)
public class CosFileSink extends BaseFileSink {

private final CatalogTable catalogTable;

public CosFileSink(CatalogTable catalogTable) {
this.catalogTable = catalogTable;
}

@Override
public String getPluginName() {
return FileSystemType.COS.getFileSystemPluginName();
Expand Down Expand Up @@ -67,6 +70,6 @@ public void prepare(Config pluginConfig) throws PrepareFailException {

@Override
public Optional<CatalogTable> getWriteCatalogTable() {
return super.getWriteCatalogTable();
return Optional.ofNullable(catalogTable);
}
}
7 changes: 7 additions & 0 deletions ...main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/sink/CosFileSinkFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@
package org.apache.seatunnel.connectors.seatunnel.file.cos.sink;

import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.table.connector.TableSink;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSinkFactory;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSinkOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
Expand Down Expand Up @@ -110,4 +112,9 @@ public OptionRule optionRule() {
.optional(FileBaseSinkOptions.TMP_PATH)
.build();
}

@Override
public TableSink createSink(TableSinkFactoryContext context) {
return () -> new CosFileSink(context.getCatalogTable());
}
}
51 changes: 18 additions & 33 deletions ...rc/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,56 +19,31 @@

import org.apache.seatunnel.shade.com.typesafe.config.Config;

import org.apache.seatunnel.api.common.PrepareFailException;
import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.common.config.CheckConfigUtil;
import org.apache.seatunnel.common.config.CheckResult;
import org.apache.seatunnel.common.constants.PluginType;
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType;
import org.apache.seatunnel.connectors.seatunnel.file.cos.config.CosConf;
import org.apache.seatunnel.connectors.seatunnel.file.cos.config.CosFileBaseOptions;
import org.apache.seatunnel.connectors.seatunnel.file.cos.config.CosFileSourceOptions;
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
import org.apache.seatunnel.connectors.seatunnel.file.source.BaseFileSource;
import org.apache.seatunnel.connectors.seatunnel.file.source.reader.ReadStrategyFactory;

import com.google.auto.service.AutoService;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

@AutoService(SeaTunnelSource.class)
public class CosFileSource extends BaseFileSource {
@Override
public String getPluginName() {
return FileSystemType.COS.getFileSystemPluginName();
}

@Override
public void prepare(Config pluginConfig) throws PrepareFailException {
CheckResult result =
CheckConfigUtil.checkAllExists(
pluginConfig,
FileBaseOptions.FILE_PATH.key(),
CosFileSourceOptions.FILE_FORMAT_TYPE.key(),
CosFileSourceOptions.SECRET_ID.key(),
CosFileSourceOptions.SECRET_KEY.key(),
CosFileSourceOptions.REGION.key(),
CosFileSourceOptions.BUCKET.key());
if (!result.isSuccess()) {
throw new FileConnectorException(
SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
String.format(
"PluginName: %s, PluginType: %s, Message: %s",
getPluginName(), PluginType.SOURCE, result.getMsg()));
}
private final CatalogTable catalogTable;

public CosFileSource(ReadonlyConfig readonlyConfig, CatalogTable catalogTable) {
this.catalogTable = catalogTable;
Config pluginConfig = readonlyConfig.toConfig();
String path = pluginConfig.getString(CosFileBaseOptions.FILE_PATH.key());
hadoopConf = CosConf.buildWithConfig(pluginConfig);
readStrategy =
Expand Down Expand Up @@ -130,4 +105,14 @@ public void prepare(Config pluginConfig) throws PrepareFailException {
}
}
}

@Override
public String getPluginName() {
return FileSystemType.COS.getFileSystemPluginName();
}

@Override
public List<CatalogTable> getProducedCatalogTables() {
return Collections.singletonList(catalogTable);
}
}
15 changes: 15 additions & 0 deletions .../java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,12 @@
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.api.table.connector.TableSource;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
Expand All @@ -30,6 +34,7 @@

import com.google.auto.service.AutoService;

import java.io.Serializable;
import java.util.Arrays;

@AutoService(Factory.class)
Expand Down Expand Up @@ -97,4 +102,14 @@ public OptionRule optionRule() {
public Class<? extends SeaTunnelSource> getSourceClass() {
return CosFileSource.class;
}

@Override
public <T, SplitT extends SourceSplit, StateT extends Serializable>
TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
return () ->
(SeaTunnelSource<T, SplitT, StateT>)
new CosFileSource(
context.getOptions(),
CatalogTableUtil.buildWithConfig(context.getOptions()));
Copy link
Member

@Hisoka-X Hisoka-X Apr 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

-1. We should build catalogtable with file sometimes. Please refer https://github.com/apache/seatunnel/pull/9121/files#diff-1f7ebddd2239c9ee909ee66c8f67c119b92f05324519ea5ce40d5aeab98da984L124

}
}
Loading