Skip to content

Commit 596e409

Browse files
ruanwenjunruanwenjun
committed
Resolve conflicts
1 parent 9c1c46e commit 596e409

File tree

10 files changed

+95
-144
lines changed

10 files changed

+95
-144
lines changed

Diff for: ‎seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/config/LocalFileSourceConfig.java

+2-80
Original file line numberDiff line numberDiff line change
@@ -18,20 +18,13 @@
1818
package org.apache.seatunnel.connectors.seatunnel.file.local.source.config;
1919

2020
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
21-
import org.apache.seatunnel.api.table.catalog.CatalogTable;
22-
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
23-
import org.apache.seatunnel.api.table.catalog.schema.TableSchemaOptions;
24-
import org.apache.seatunnel.common.constants.PluginType;
25-
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
21+
import org.apache.seatunnel.connectors.seatunnel.file.config.BaseFileSourceConfig;
2622
import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType;
2723
import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
2824
import org.apache.seatunnel.connectors.seatunnel.file.local.config.LocalFileHadoopConf;
2925

3026
import lombok.Getter;
3127

32-
import java.io.Serializable;
33-
import java.util.List;
34-
3528
@Getter
3629
public class LocalFileSourceConfig extends BaseFileSourceConfig {
3730

@@ -48,77 +41,6 @@ public String getPluginName() {
4841
}
4942

5043
public LocalFileSourceConfig(ReadonlyConfig readonlyConfig) {
51-
validateConfig(readonlyConfig);
52-
this.fileFormat = readonlyConfig.get(LocalFileSourceOptions.FILE_FORMAT_TYPE);
53-
this.localFileHadoopConf = new LocalFileHadoopConf();
54-
this.readStrategy = ReadStrategyFactory.of(readonlyConfig, localFileHadoopConf);
55-
this.filePaths = parseFilePaths(readonlyConfig);
56-
this.catalogTable = parseCatalogTable(readonlyConfig);
57-
}
58-
59-
private void validateConfig(ReadonlyConfig readonlyConfig) {
60-
if (!readonlyConfig.getOptional(LocalFileSourceOptions.FILE_PATH).isPresent()) {
61-
throw new FileConnectorException(
62-
SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
63-
String.format(
64-
"PluginName: %s, PluginType: %s, Message: %s",
65-
FileSystemType.LOCAL.getFileSystemPluginName(),
66-
PluginType.SOURCE,
67-
LocalFileSourceOptions.FILE_PATH + " is required"));
68-
}
69-
if (!readonlyConfig.getOptional(LocalFileSourceOptions.FILE_FORMAT_TYPE).isPresent()) {
70-
throw new FileConnectorException(
71-
SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
72-
String.format(
73-
"PluginName: %s, PluginType: %s, Message: %s",
74-
FileSystemType.LOCAL.getFileSystemPluginName(),
75-
PluginType.SOURCE,
76-
LocalFileSourceOptions.FILE_FORMAT_TYPE.key() + " is required"));
77-
}
78-
}
79-
80-
private List<String> parseFilePaths(ReadonlyConfig readonlyConfig) {
81-
String rootPath = null;
82-
try {
83-
rootPath = readonlyConfig.get(LocalFileSourceOptions.FILE_PATH);
84-
return readStrategy.getFileNamesByPath(localFileHadoopConf, rootPath);
85-
} catch (Exception ex) {
86-
String errorMsg = String.format("Get file list from this path [%s] failed", rootPath);
87-
throw new FileConnectorException(
88-
FileConnectorErrorCode.FILE_LIST_GET_FAILED, errorMsg, ex);
89-
}
90-
}
91-
92-
private CatalogTable parseCatalogTable(ReadonlyConfig readonlyConfig) {
93-
final CatalogTable catalogTable;
94-
if (readonlyConfig.getOptional(TableSchemaOptions.SCHEMA).isPresent()) {
95-
catalogTable =
96-
CatalogTableUtil.buildWithConfig(
97-
FileSystemType.LOCAL.getFileSystemPluginName(), readonlyConfig);
98-
} else {
99-
catalogTable = CatalogTableUtil.buildSimpleTextTable();
100-
}
101-
if (CollectionUtils.isEmpty(filePaths)) {
102-
return catalogTable;
103-
}
104-
switch (fileFormat) {
105-
case CSV:
106-
case TEXT:
107-
case JSON:
108-
case EXCEL:
109-
readStrategy.setSeaTunnelRowTypeInfo(catalogTable.getSeaTunnelRowType());
110-
return CatalogTableUtil.newCatalogTable(
111-
catalogTable, readStrategy.getActualSeaTunnelRowTypeInfo());
112-
case ORC:
113-
case PARQUET:
114-
return CatalogTableUtil.newCatalogTable(
115-
catalogTable,
116-
readStrategy.getSeaTunnelRowTypeInfo(
117-
localFileHadoopConf, filePaths.get(0)));
118-
default:
119-
throw new FileConnectorException(
120-
FileConnectorErrorCode.FORMAT_NOT_SUPPORT,
121-
"SeaTunnel does not supported this file format: [" + fileFormat + "]");
122-
}
44+
super(readonlyConfig);
12345
}
12446
}

Diff for: ‎seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/commit/HiveSinkAggregatedCommitter.java

+3-4
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,12 @@
1818
package org.apache.seatunnel.connectors.seatunnel.hive.commit;
1919

2020
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
21+
import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
2122
import org.apache.seatunnel.connectors.seatunnel.file.sink.commit.FileAggregatedCommitInfo;
2223
import org.apache.seatunnel.connectors.seatunnel.file.sink.commit.FileSinkAggregatedCommitter;
23-
import org.apache.seatunnel.connectors.seatunnel.file.sink.util.FileSystemUtils;
2424
import org.apache.seatunnel.connectors.seatunnel.hive.sink.HiveSinkOptions;
2525
import org.apache.seatunnel.connectors.seatunnel.hive.utils.HiveMetaStoreProxy;
2626

27-
import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
2827
import org.apache.hadoop.hive.metastore.api.Table;
2928
import org.apache.thrift.TException;
3029

@@ -44,8 +43,8 @@ public class HiveSinkAggregatedCommitter extends FileSinkAggregatedCommitter {
4443
private final ReadonlyConfig readonlyConfig;
4544

4645
public HiveSinkAggregatedCommitter(
47-
ReadonlyConfig readonlyConfig, Table table, FileSystemUtils fileSystemUtils) {
48-
super(fileSystemUtils);
46+
ReadonlyConfig readonlyConfig, Table table, HadoopConf hadoopConf) {
47+
super(hadoopConf);
4948
this.readonlyConfig = readonlyConfig;
5049
this.dbName = table.getDbName();
5150
this.tableName = table.getTableName();

Diff for: ‎seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/config/BaseHiveOptions.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,9 @@
1919

2020
import org.apache.seatunnel.api.configuration.Option;
2121
import org.apache.seatunnel.api.configuration.Options;
22-
import org.apache.seatunnel.connectors.seatunnel.file.config.BaseSourceConfig;
22+
import org.apache.seatunnel.connectors.seatunnel.file.config.BaseSourceConfigOptions;
2323

24-
public class BaseHiveOptions extends BaseSourceConfig {
24+
public class BaseHiveOptions extends BaseSourceConfigOptions {
2525

2626
public static final Option<String> TABLE_NAME =
2727
Options.key("table_name")

Diff for: ‎seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/sink/HiveSink.java

+2-5
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636
import org.apache.seatunnel.connectors.seatunnel.file.sink.commit.FileCommitInfo;
3737
import org.apache.seatunnel.connectors.seatunnel.file.sink.config.FileSinkConfig;
3838
import org.apache.seatunnel.connectors.seatunnel.file.sink.state.FileSinkState;
39-
import org.apache.seatunnel.connectors.seatunnel.file.sink.util.FileSystemUtils;
4039
import org.apache.seatunnel.connectors.seatunnel.file.sink.writer.WriteStrategy;
4140
import org.apache.seatunnel.connectors.seatunnel.file.sink.writer.WriteStrategyFactory;
4241
import org.apache.seatunnel.connectors.seatunnel.hive.commit.HiveSinkAggregatedCommitter;
@@ -75,7 +74,6 @@ public class HiveSink
7574
private final CatalogTable catalogTable;
7675
private final ReadonlyConfig readonlyConfig;
7776
private final HiveHadoopConfig hiveHadoopConfig;
78-
private final FileSystemUtils fileSystemUtils;
7977
private final FileSinkConfig fileSinkConfig;
8078
private final WriteStrategy writeStrategy;
8179
private String jobId;
@@ -89,9 +87,7 @@ public HiveSink(ReadonlyConfig readonlyConfig, CatalogTable catalogTable) {
8987
generateFileSinkConfig(readonlyConfig, tableInformation, catalogTable);
9088
this.writeStrategy =
9189
WriteStrategyFactory.of(fileSinkConfig.getFileFormat(), fileSinkConfig);
92-
this.fileSystemUtils = new FileSystemUtils(hiveHadoopConfig);
9390
this.writeStrategy.setSeaTunnelRowTypeInfo(catalogTable.getSeaTunnelRowType());
94-
this.writeStrategy.setFileSystemUtils(fileSystemUtils);
9591
}
9692

9793
private FileSinkConfig generateFileSinkConfig(
@@ -171,7 +167,8 @@ public String getPluginName() {
171167
public Optional<SinkAggregatedCommitter<FileCommitInfo, FileAggregatedCommitInfo>>
172168
createAggregatedCommitter() {
173169
return Optional.of(
174-
new HiveSinkAggregatedCommitter(readonlyConfig, tableInformation, fileSystemUtils));
170+
new HiveSinkAggregatedCommitter(
171+
readonlyConfig, tableInformation, hiveHadoopConfig));
175172
}
176173

177174
@Override

Diff for: ‎seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/sink/HiveSinkFactory.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@
4141
@AutoService(Factory.class)
4242
public class HiveSinkFactory
4343
implements TableSinkFactory<
44-
SeaTunnelRow, FileSinkState, FileCommitInfo, FileAggregatedCommitInfo> {
44+
SeaTunnelRow, FileSinkState, FileCommitInfo, FileAggregatedCommitInfo> {
4545

4646
@Override
4747
public OptionRule optionRule() {
@@ -55,7 +55,7 @@ public OptionRule optionRule() {
5555

5656
@Override
5757
public TableSink<SeaTunnelRow, FileSinkState, FileCommitInfo, FileAggregatedCommitInfo>
58-
createSink(TableSinkFactoryContext context) {
58+
createSink(TableSinkFactoryContext context) {
5959
ReadonlyConfig readonlyConfig = context.getOptions();
6060
CatalogTable catalogTable = context.getCatalogTable();
6161

@@ -80,7 +80,7 @@ private ReadonlyConfig generateCurrentReadonlyConfig(
8080
tableName -> {
8181
String replacedPath =
8282
replaceCatalogTableInPath(tableName, catalogTable);
83-
configMap.put(BaseSinkConfig.FILE_PATH.key(), replacedPath);
83+
configMap.put(HiveSinkOptions.TABLE_NAME.key(), replacedPath);
8484
});
8585

8686
return ReadonlyConfig.fromMap(new HashMap<>(configMap));

Diff for: ‎seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/source/HiveSourceFactory.java

+9-9
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,7 @@
2424
import org.apache.seatunnel.api.table.factory.Factory;
2525
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
2626
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
27-
import org.apache.seatunnel.connectors.seatunnel.file.config.BaseSourceConfig;
28-
import org.apache.seatunnel.connectors.seatunnel.file.hdfs.source.config.HdfsSourceConfig;
27+
import org.apache.seatunnel.connectors.seatunnel.file.hdfs.source.config.HdfsSourceConfigOptions;
2928
import org.apache.seatunnel.connectors.seatunnel.hive.config.HiveConstants;
3029
import org.apache.seatunnel.connectors.seatunnel.hive.source.config.HiveSourceOptions;
3130

@@ -49,16 +48,17 @@ TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
4948
@Override
5049
public OptionRule optionRule() {
5150
return OptionRule.builder()
52-
.required(HiveSourceOptions.TABLE_NAME)
53-
.required(HiveSourceOptions.METASTORE_URI)
51+
.optional(HiveSourceOptions.TABLE_NAME)
52+
.optional(HiveSourceOptions.METASTORE_URI)
5453
.optional(HiveSourceOptions.TABLE_CONFIGS)
5554
.optional(HiveSourceOptions.HIVE_SITE_PATH)
5655
.optional(
57-
HdfsSourceConfig.HDFS_SITE_PATH,
58-
HdfsSourceConfig.KERBEROS_PRINCIPAL,
59-
HdfsSourceConfig.KERBEROS_KEYTAB_PATH)
60-
.optional(BaseSourceConfig.READ_PARTITIONS)
61-
.optional(BaseSourceConfig.READ_COLUMNS)
56+
HdfsSourceConfigOptions.HDFS_SITE_PATH,
57+
HdfsSourceConfigOptions.REMOTE_USER,
58+
HdfsSourceConfigOptions.KERBEROS_PRINCIPAL,
59+
HdfsSourceConfigOptions.KERBEROS_KEYTAB_PATH)
60+
.optional(HdfsSourceConfigOptions.READ_PARTITIONS)
61+
.optional(HdfsSourceConfigOptions.READ_COLUMNS)
6262
.build();
6363
}
6464

Diff for: ‎seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/source/config/HiveSourceConfig.java

+56-13
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,9 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.hive.source.config;
1919

20+
import org.apache.seatunnel.shade.com.typesafe.config.Config;
21+
import org.apache.seatunnel.shade.com.typesafe.config.ConfigValueFactory;
22+
2023
import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
2124
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
2225
import org.apache.seatunnel.api.table.catalog.CatalogTable;
@@ -28,11 +31,10 @@
2831
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
2932
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
3033
import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
31-
import org.apache.seatunnel.connectors.seatunnel.file.config.BaseSourceConfig;
3234
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
3335
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
3436
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
35-
import org.apache.seatunnel.connectors.seatunnel.file.hdfs.source.config.HdfsSourceConfig;
37+
import org.apache.seatunnel.connectors.seatunnel.file.hdfs.source.config.HdfsSourceConfigOptions;
3638
import org.apache.seatunnel.connectors.seatunnel.file.source.reader.ReadStrategy;
3739
import org.apache.seatunnel.connectors.seatunnel.file.source.reader.ReadStrategyFactory;
3840
import org.apache.seatunnel.connectors.seatunnel.hive.config.HiveConstants;
@@ -55,6 +57,11 @@
5557
import java.util.ArrayList;
5658
import java.util.HashMap;
5759
import java.util.List;
60+
import java.util.Map;
61+
62+
import static org.apache.seatunnel.connectors.seatunnel.file.config.BaseSinkConfig.FIELD_DELIMITER;
63+
import static org.apache.seatunnel.connectors.seatunnel.file.config.BaseSinkConfig.FILE_FORMAT_TYPE;
64+
import static org.apache.seatunnel.connectors.seatunnel.file.config.BaseSinkConfig.ROW_DELIMITER;
5865

5966
@Getter
6067
public class HiveSourceConfig implements Serializable {
@@ -71,13 +78,13 @@ public class HiveSourceConfig implements Serializable {
7178
@SneakyThrows
7279
public HiveSourceConfig(ReadonlyConfig readonlyConfig) {
7380
readonlyConfig
74-
.getOptional(BaseSourceConfig.READ_PARTITIONS)
81+
.getOptional(HdfsSourceConfigOptions.READ_PARTITIONS)
7582
.ifPresent(this::validatePartitions);
7683
this.table = HiveTableUtils.getTableInfo(readonlyConfig);
7784
this.hiveHadoopConfig = parseHiveHadoopConfig(readonlyConfig, table);
7885
this.fileFormat = HiveTableUtils.parseFileFormat(table);
79-
this.readStrategy = parseReadStrategy(readonlyConfig, fileFormat, hiveHadoopConfig);
80-
this.filePaths = parseFilePaths(table, hiveHadoopConfig, readStrategy);
86+
this.readStrategy = parseReadStrategy(table, readonlyConfig, fileFormat, hiveHadoopConfig);
87+
this.filePaths = parseFilePaths(table, readStrategy);
8188
this.catalogTable =
8289
parseCatalogTable(
8390
readonlyConfig,
@@ -108,11 +115,45 @@ private void validatePartitions(List<String> partitionsList) {
108115
}
109116

110117
private ReadStrategy parseReadStrategy(
118+
Table table,
111119
ReadonlyConfig readonlyConfig,
112120
FileFormat fileFormat,
113121
HiveHadoopConfig hiveHadoopConfig) {
122+
114123
ReadStrategy readStrategy = ReadStrategyFactory.of(fileFormat.name());
115-
readStrategy.setPluginConfig(readonlyConfig.toConfig());
124+
Config config = readonlyConfig.toConfig();
125+
126+
switch (fileFormat) {
127+
case TEXT:
128+
// if the file format is text, we set the delim.
129+
Map<String, String> parameters = table.getSd().getSerdeInfo().getParameters();
130+
config =
131+
config.withValue(
132+
FIELD_DELIMITER.key(),
133+
ConfigValueFactory.fromAnyRef(
134+
parameters.get("field.delim")))
135+
.withValue(
136+
ROW_DELIMITER.key(),
137+
ConfigValueFactory.fromAnyRef(parameters.get("line.delim")))
138+
.withValue(
139+
FILE_FORMAT_TYPE.key(),
140+
ConfigValueFactory.fromAnyRef(FileFormat.TEXT.name()));
141+
break;
142+
case ORC:
143+
config =
144+
config.withValue(
145+
FILE_FORMAT_TYPE.key(),
146+
ConfigValueFactory.fromAnyRef(FileFormat.ORC.name()));
147+
break;
148+
case PARQUET:
149+
config =
150+
config.withValue(
151+
FILE_FORMAT_TYPE.key(),
152+
ConfigValueFactory.fromAnyRef(FileFormat.PARQUET.name()));
153+
break;
154+
default:
155+
}
156+
readStrategy.setPluginConfig(config);
116157
readStrategy.init(hiveHadoopConfig);
117158
return readStrategy;
118159
}
@@ -125,22 +166,24 @@ private HiveHadoopConfig parseHiveHadoopConfig(ReadonlyConfig readonlyConfig, Ta
125166
readonlyConfig.get(HiveSourceOptions.METASTORE_URI),
126167
readonlyConfig.get(HiveSourceOptions.HIVE_SITE_PATH));
127168
readonlyConfig
128-
.getOptional(HdfsSourceConfig.HDFS_SITE_PATH)
169+
.getOptional(HdfsSourceConfigOptions.HDFS_SITE_PATH)
129170
.ifPresent(hiveHadoopConfig::setHdfsSitePath);
130171
readonlyConfig
131-
.getOptional(HdfsSourceConfig.KERBEROS_PRINCIPAL)
172+
.getOptional(HdfsSourceConfigOptions.KERBEROS_PRINCIPAL)
132173
.ifPresent(hiveHadoopConfig::setKerberosPrincipal);
133174
readonlyConfig
134-
.getOptional(HdfsSourceConfig.KERBEROS_KEYTAB_PATH)
175+
.getOptional(HdfsSourceConfigOptions.KERBEROS_KEYTAB_PATH)
135176
.ifPresent(hiveHadoopConfig::setKerberosKeytabPath);
177+
readonlyConfig
178+
.getOptional(HdfsSourceConfigOptions.REMOTE_USER)
179+
.ifPresent(hiveHadoopConfig::setRemoteUser);
136180
return hiveHadoopConfig;
137181
}
138182

139-
private List<String> parseFilePaths(
140-
Table table, HiveHadoopConfig hiveHadoopConfig, ReadStrategy readStrategy) {
183+
private List<String> parseFilePaths(Table table, ReadStrategy readStrategy) {
141184
String hdfsPath = parseHdfsPath(table);
142185
try {
143-
return readStrategy.getFileNamesByPath(hiveHadoopConfig, hdfsPath);
186+
return readStrategy.getFileNamesByPath(hdfsPath);
144187
} catch (Exception e) {
145188
String errorMsg = String.format("Get file list from this path [%s] failed", hdfsPath);
146189
throw new FileConnectorException(
@@ -214,7 +257,7 @@ private CatalogTable parseCatalogTableFromRemotePath(
214257
CatalogTable catalogTable = buildEmptyCatalogTable(readonlyConfig, table);
215258
try {
216259
SeaTunnelRowType seaTunnelRowTypeInfo =
217-
readStrategy.getSeaTunnelRowTypeInfo(hiveHadoopConfig, filePaths.get(0));
260+
readStrategy.getSeaTunnelRowTypeInfo(filePaths.get(0));
218261
return CatalogTableUtil.newCatalogTable(catalogTable, seaTunnelRowTypeInfo);
219262
} catch (FileConnectorException e) {
220263
String errorMsg =

0 commit comments

Comments
 (0)