[Feature][Spark] Support multiple tables read and write by Carl-Zhou-CN · Pull Request #7283 · apache/seatunnel

Carl-Zhou-CN · 2024-07-29T08:27:46Z

close #7009

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update change log that in connector document. For more details you can refer to connector-v2
2. Update plugin-mapping.properties and add new connector information in it
3. Update the pom file of seatunnel-dist
Update the release-note.

hailin0 · 2024-08-01T04:02:45Z

cc @CheneyYin

CheneyYin · 2024-08-01T04:48:58Z

            this.jobName = config.getString("job.name");
        }
        sparkConf = createSparkConf();
+        sparkConf.set("spark.sql.unsafe.enabled", "false");


why disable unsafe?

I'll remove it. Thank you

CheneyYin · 2024-08-01T05:42:23Z

    public InternalRow convert(SeaTunnelRow seaTunnelRow) throws IOException {
-        validate(seaTunnelRow);
-        return (InternalRow) convert(seaTunnelRow, dataType);
+        return parcel(seaTunnelRow, (SeaTunnelRowType) dataType);
    }


It might be more reliable to add unit tests for it. It would be best if it covers all data types.

I will refactor and consolidate this part of the logic and provide test cases.

CheneyYin · 2024-08-01T05:48:06Z

    public SeaTunnelRow reconvert(InternalRow engineRow) throws IOException {
-        return (SeaTunnelRow) reconvert(engineRow, dataType);
+        return unpack(engineRow, (SeaTunnelRowType) dataType);
    }


CheneyYin · 2024-08-01T05:50:05Z

        return newRow;
    }

+    public GenericRowWithSchema parcel(SeaTunnelRow seaTunnelRow) {


CheneyYin · 2024-08-01T05:50:17Z

        return (SeaTunnelRow) reconvert(engineRow, dataType);
    }

+    public SeaTunnelRow unpack(GenericRowWithSchema engineRow) throws IOException {


CheneyYin · 2024-08-01T06:19:52Z

+        List<String> fieldNames = new ArrayList<>();
+        List<SeaTunnelDataType<?>> fieldTypes = new ArrayList<>();
+        List<ColumnWithIndex> columnWithIndexes = new ArrayList<>();
+        fieldTypes.sort(Comparator.comparingInt(o -> o.getSqlType().ordinal()));


fieldTypes is empty. It's meaningless to sort it.

It's not being used. I'll remove it. Thank you

CheneyYin · 2024-08-01T06:32:23Z

+
+public class SchemaUtil {
+
+    public static ColumnWithIndex[] mergeSchema(List<CatalogTable> catalogTables) {


It would be better to add unit test cases.

hailin0 · 2024-08-10T05:47:51Z

waiting for this pr merge
#7356

hailin0

Please add a screenshot of the log for multi-table mapping

hailin0 · 2024-08-10T05:50:04Z

+        //  if (catalogTables.size() != 1) {
+        //      throw new SeaTunnelException(
+        //              String.format("Unsupported table number: %d on flink",
+        // catalogTables.size()));
+        //  }


This is specifically for multi-table exceptions and can be removed when both spark and flink are implemented

flink is currently not supported

It affects not only flink but also spark

So flink will not throw exception now?

yes, I think flink is about to implement multiple tables as well

Ok. Please clean the code.

Let's clean it up when flink implements multiple tables

Carl-Zhou-CN · 2024-08-10T06:21:22Z

CheneyYin · 2024-08-10T12:11:35Z

        }
        this.sparkSession = builder.getOrCreate();
        createStreamingContext();
+


done, good fix

CheneyYin · 2024-08-10T12:22:30Z

                TableTransformFactoryContext context =
                        new TableTransformFactoryContext(
-                                Collections.singletonList(dataset.getCatalogTable()),
+                                Collections.singletonList(dataset.getCatalogTables().get(0)),


Maybe you can use dataset.getCatalogTables() instead of Collections.singletonList(dataset.getCatalogTables().get(0)) ?. Like this:

TableTransformFactoryContext context = new TableTransformFactoryContext( dataset.getCatalogTables(), ReadonlyConfig.fromConfig(pluginConfig), classLoader);

done, excellent advice

CheneyYin · 2024-08-10T12:28:39Z

        Dataset<Row> stream = tableInfo.getDataset();
-        SeaTunnelDataType<?> inputDataType = tableInfo.getCatalogTable().getSeaTunnelRowType();
+        SeaTunnelDataType<?> inputDataType =
+                tableInfo.getCatalogTables().get(0).getSeaTunnelRowType();


Does Covenant transform only deal with the first one?

This is because 'transform' now supports only single tables

CheneyYin

LGTM

jackyyyyyssss · 2024-08-12T08:55:45Z

                mockServerClient.retrieveRecordedRequests(
                        request().withPath("/example/httpMultiTableContentSink").withMethod("POST"),
                        Format.JSON);
+        mockServerClient.clear(


Sorry, I forgot to clear the records, which caused the run to fail. This test case will run multiple times. Please help add it

Thank you very much @jackyyyyyssss ,Please cc

hailin0

LGTM

cc @Hisoka-X

github-actions Bot added core SeaTunnel core module Spark format labels Jul 29, 2024

Carl-Zhou-CN added 2 commits July 29, 2024 16:40

[Feature][Spark] Support multiple tables read and write

e1a8d65

[Feature][Spark] Support multiple tables read and write

73e3bb1

Carl-Zhou-CN force-pushed the spark-rewrite branch from 5976900 to 73e3bb1 Compare July 29, 2024 09:11

Carl-Zhou-CN added 3 commits July 29, 2024 17:19

[Feature][Spark] Support multiple tables read and write

177437e

[Feature][Spark] Support multiple tables read and write

bec7e4d

[Feature][Spark] Support multiple tables read and write

4ddc11d

github-actions Bot removed the format label Jul 29, 2024

[Feature][Spark] Support multiple tables read and write

6dba882

github-actions Bot added the e2e label Jul 29, 2024

hailin0 reviewed Jul 29, 2024

View reviewed changes

Comment thread .../src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlMultipleTablesIT.java

[Feature][Spark] Support multiple tables read and write

b9bcf56

CheneyYin reviewed Aug 1, 2024

View reviewed changes

Carl-Zhou-CN added 2 commits August 6, 2024 22:22

[Feature][Spark] Support multiple tables read and write

3ff74c9

[Feature][Spark] Support multiple tables read and write

d4dff63

Carl-Zhou-CN requested review from CheneyYin and hailin0 August 7, 2024 02:41

hailin0 reviewed Aug 7, 2024

View reviewed changes

Comment thread .../src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlMultipleTablesIT.java

Carl-Zhou-CN added 3 commits August 8, 2024 10:27

[Feature][Spark] Support multiple tables read and write

823d68a

[Feature][Spark] Support multiple tables read and write

5239276

[Feature][Spark] Support multiple tables read and write

d40f5f5

Carl-Zhou-CN mentioned this pull request Aug 8, 2024

[Feature][e2e] testMultiTableHttp test case is not suitable for spark or flink #7345

Closed

3 tasks

hailin0 reviewed Aug 10, 2024

View reviewed changes

CheneyYin reviewed Aug 10, 2024

View reviewed changes

fix

5e74b18

Carl-Zhou-CN requested a review from CheneyYin August 11, 2024 04:59

CheneyYin reviewed Aug 11, 2024

View reviewed changes

Carl-Zhou-CN requested a review from hailin0 August 12, 2024 01:57

Carl-Zhou-CN and others added 3 commits August 12, 2024 14:04

Merge branch 'apache:dev' into spark-rewrite

f781c99

fix

6a20570

fix

efbd5a8

jackyyyyyssss reviewed Aug 12, 2024

View reviewed changes

Carl-Zhou-CN requested a review from jackyyyyyssss August 12, 2024 08:58

CheneyYin approved these changes Aug 12, 2024

View reviewed changes

github-actions Bot added the reviewed label Aug 12, 2024

hailin0 reviewed Aug 12, 2024

View reviewed changes

hailin0 approved these changes Aug 12, 2024

View reviewed changes

github-actions Bot added the approved label Aug 12, 2024

Hisoka-X approved these changes Aug 14, 2024

View reviewed changes

Hisoka-X merged commit ff107ac into apache:dev Aug 14, 2024

Hisoka-X mentioned this pull request Aug 15, 2024

[Improve][Connector-V2]Support multi-table sink feature for email #7368

Merged

hailin0 mentioned this pull request Aug 26, 2024

SeaTunnel community meeting Topic collect #7499

Closed

3 tasks


		public class SchemaUtil {

		public static ColumnWithIndex[] mergeSchema(List<CatalogTable> catalogTables) {

Uh oh!

Conversation

Carl-Zhou-CN commented Jul 29, 2024

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

Uh oh!

Uh oh!

hailin0 commented Aug 1, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

hailin0 commented Aug 10, 2024

Uh oh!

hailin0 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Carl-Zhou-CN commented Aug 10, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Carl-Zhou-CN Aug 11, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

CheneyYin left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hailin0 left a comment

Choose a reason for hiding this comment

Carl-Zhou-CN Aug 11, 2024 •

edited

Loading