[Feature][Connector-V2] Support the MySQL compatibility mode of KingbaseES

[PZh101]

Under the MySQL compatibility mode of KingbaseES, adopt MySQL type mapping, row converters and Upsert statements, and dynamically initialize the compatibility level based on connections. Meanwhile, fix the issue that the specified fetchSize does not take effect in MysqlDialect.

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 necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[DanielLeens]

Thanks for the contribution. I went through the full latest diff and traced the actual JDBC source/sink call paths rather than only the metadata path. The direction makes sense, but the current implementation still does not make KingbaseES MySQL compatibility work on the real runtime path.

What the PR is trying to solve:

• User pain: when KingbaseES runs in MySQL compatibility mode, users expect MySQL-style quoting, row conversion, and read/write behavior without getting mismatched dialect semantics.
• Fix approach: the PR adds a compatibleLevel to KingbaseDialect and tries to initialize it from KbConnection.getCompatibleLevel().
• One-line summary: the automatic detection is only wired into one metadata fallback path today, not into the actual source reader / sink writer runtime path.

Runtime chain I checked:

Source init
  -> JdbcInputFormat(config, tables)
      -> JdbcDialectLoader.load(...)
      -> jdbcRowConverter = jdbcDialect.getRowConverter()
      -> no connection-based compatibility init has happened yet

Catalog fallback
  -> JdbcCatalogUtils.getTables(...)
      -> getConnection(...)
      -> jdbcDialect.initCompatibleByConnection(connection)
      -> getCatalogTable(...)

Sink init
  -> JdbcSinkFactory.createSink(...)
      -> JdbcDialectLoader.load(...)
      -> dialect.connectionUrlParse(...)
      -> new JdbcSink(...)
      -> no initCompatibleByConnection(connection) here either

Key files:

• seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java:67-75
• seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/utils/JdbcCatalogUtils.java:143-147
• seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactory.java:180-194
• seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/kingbase/KingbaseDialect.java:37-78

Issue 1: connection-based compatibility detection does not reach the real source/sink runtime path

• Location: seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java:67
• Why this is a blocker:
  JdbcInputFormat chooses the row converter during construction, before any connection-based initialization can happen. The sink path also never calls initCompatibleByConnection(...). So if a user relies on this PR instead of explicitly setting compatible_mode=mysql, the latest head still behaves like the default Kingbase dialect on the real read/write path.
• Risk:
  the PR would claim support that is only partially implemented and can still fail on the main user path.
• Better fix:
  determine the compatibility mode in one shared place before constructing the source/sink dialect, and then pass that result in as an immutable input.
• Severity: High

Issue 2: the PR introduces mutable dialect state without closing the contract

• Location: seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/kingbase/KingbaseDialect.java:37
• Why this matters:
  JdbcDialect explicitly documents dialects as immutable and stateless, but the new implementation adds connection-discovered mutable state and does not define who is responsible for initializing it on each path.
• Severity: Medium

Compatibility:

• Existing jobs that explicitly set compatible_mode=mysql remain okay.
• The new “auto-detect from connection” behavior is not complete yet across source, sink, and catalog paths.

Tests:

• I did not see new runtime regression coverage for the exact path above.
• That is why the current gap can slip through even though the metadata fallback code itself looks reasonable.

CI:

• I did not run this locally.
• The current Build failure is Run / updated-modules-integration-test-part-1 (8, ubuntu-latest) on the fork workflow.
• The failing test is org.apache.seatunnel.connectors.jdbc.SqlServerSchemaChangeIT, and the log shows Status 409: ... container ... is not running, followed by an Awaitility timeout expecting 9 rows but getting 0.
• That failure does not line up with the Kingbase compatibility path changed here, so I would treat it as separate CI noise rather than evidence that the current blocker is solved.

Merge conclusion: can merge after fixes

1. Blocking items

• Issue 1 must be fixed first, because the feature is not actually wired into the main source/sink runtime path yet.

2. Suggested follow-up

• Clarify the dialect lifecycle/immutability contract from Issue 2, and add a runtime regression test once the main path is wired through.

Overall, the idea is useful, but the latest head still stops one step too early: it improves metadata fallback behavior, not the full KingbaseES MySQL-compatibility runtime story.

[PZh101]

Appreciate your careful review! I've understood the two core issues clearly.
I will fix the runtime path compatibility detection first, unify the compatibility mode initialization logic for source and sink, and refactor the dialect to keep stateless and immutable. Meanwhile, I will supplement corresponding runtime tests.
I'll update the PR shortly and look forward to your recheck.

[DanielLeens]

Thanks for the quick follow-up, and I appreciate the clear plan. That sounds like the right direction: first make the compatibility detection reach the real source/sink runtime path, then keep the dialect initialization logic consistent and stateless. Since there is still no new commit after the latest Daniel review yet, I am not starting another full re-review on the current revision in this round. Please ping me again once the code update is pushed, and I will recheck the new head from scratch.

[PZh101]

Hello, I've made the adjustments according to your feedback. Could you please review them? Thank you.

[DanielLeens]

Thanks for the update and for addressing the two core blockers from the previous round. I pulled the latest head 1c7fdcc40d69c573b2bf52d0f396a1dfe658221d locally and re-reviewed the current JDBC source / sink / catalog call paths from scratch.

The good news first: the latest head does fix the two main issues I called out before. The compatibility-mode detection is now wired into the real runtime construction path, not just the catalog fallback path, and the Kingbase dialect state is now fixed at construction time instead of being left mutable.

That said, I still found one blocking issue in the current revision, plus one documentation gap.

What this PR fixes

• User pain: KingbaseES running in MySQL compatibility mode should behave like a MySQL-compatible JDBC source/sink/catalog path instead of silently falling back to the default Kingbase dialect behavior.
• Fix approach: pass JdbcConnectionConfig down into the dialect loader, detect compatibility mode from the real Kingbase connection when needed, and build an immutable KingbaseDialect with that result.
• One-line summary: the Kingbase runtime wiring is much better now, but the PR also changes a real MySQL runtime path that is outside the target feature.

Runtime chain I checked

Kingbase source init
  -> JdbcInputFormat(config, tables)
      -> JdbcDialectLoader.load(url, dialect, compatibleMode, jdbcConnectionConfig)
          -> KingbaseDialectFactory.create(...)
              -> detectCompatibleMode(config) when compatible_mode is not set
              -> new KingbaseDialect(detectedCompatibleMode, fieldIde)
      -> jdbcRowConverter = jdbcDialect.getRowConverter()

Kingbase sink init
  -> JdbcSinkFactory.createSink(...)
      -> JdbcDialectLoader.load(..., jdbcConnectionConfig)
      -> dialect.getUpsertStatement(...) / dialect.getRowConverter()

Catalog path
  -> JdbcCatalogUtils.getCatalog(...)
      -> JdbcDialectLoader.load(..., jdbcConnectionConfig)
      -> dialect.getJdbcDialectTypeMapper()

Extra MySQL path affected by this PR
  -> JdbcInputFormat.open(...)
      -> chunkSplitter.generateSplitStatement(...)
          -> ChunkSplitter.createPreparedStatement(...)
              -> jdbcDialect.creatPreparedStatement(connection, sql, fetchSize)
                  -> MysqlDialect.creatPreparedStatement(...)

Findings

Problem 1: MysqlDialect.creatPreparedStatement(...) changes the real MySQL source behavior even though it is not part of the Kingbase fix

• Location:

  • seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlDialect.java:121-130
  • seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/ChunkSplitter.java:139-147
  • seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java:96-103

• Severity: High

• Raised before by others: No

• Why this is a real blocker:
  The Kingbase wiring itself is now much closer to correct, but this extra MySQL change is on a separate live runtime path. JdbcInputFormat.open(...) creates the split statement through ChunkSplitter.createPreparedStatement(...), which delegates directly to jdbcDialect.creatPreparedStatement(...). For a real MySQL job, the old code always forced Integer.MIN_VALUE, while the new code switches to setFetchSize(fetchSize) whenever fetch_size > 0.

  That means this PR is not only fixing KingbaseES MySQL compatibility. It is also changing the prepared-statement behavior of existing MySQL source jobs. And importantly, this extra change does not help the Kingbase URL path, because Kingbase still resolves to KingbaseDialect, not MysqlDialect.

• Risk:
  this can change the fetch / memory behavior of real MySQL source jobs that already rely on the current streaming-oriented path.

• Better fix:
  Please keep the Kingbase runtime wiring changes, but revert this MysqlDialect behavior change from the PR. If Kingbase MySQL compatibility really needs a different fetch strategy, it should be implemented in the Kingbase-specific path instead of changing the live MySQL dialect behavior.

Problem 2: the user-visible Kingbase compatibility feature still needs docs in both docs/en and docs/zh

• Location:
  • docs/en/connectors/source/Kingbase.md
  • docs/en/connectors/sink/Kingbase.md
  • docs/zh/connectors/source/Kingbase.md
  • docs/zh/connectors/sink/Kingbase.md
• Severity: Medium
• Raised before by others: No
• Why this matters:
  this PR adds a user-facing capability, but the Kingbase connector docs still do not explain how compatible_mode=mysql and the connection-based detection are supposed to be used.
• Better fix:
  add a short section describing when users should set compatible_mode=mysql, when the connection-based detection is expected to help, and which behaviors are affected.

Testing / stability / CI

• The new Kingbase unit tests look structurally stable to me. They are deterministic constructor / behavior checks and I do not see obvious flaky-test patterns in the new test code.
• I did not run this locally.
• The current GitHub Build check is red. From the check metadata I could confirm the failing fork workflow jobs are unit-test (8) and updated-modules-integration-test-part-1 (11), but this run did not expose the detailed failing test text through the GitHub API output I could retrieve here. So I cannot honestly attribute the current CI red directly to the Kingbase path from metadata alone.
• Also, the current mergeStateStatus=BLOCKED is the review gate, not a merge conflict.

Conclusion: can merge after fixes

1. Blocking items

• Problem 1 must be fixed first. The Kingbase runtime wiring looks much better now, but the PR still changes a real MySQL source path that is outside the feature being added.

2. Suggested follow-ups

• Problem 2: please add the Kingbase docs updates in both English and Chinese.

Overall, the main Kingbase direction is now close. If you keep the Kingbase-specific runtime fix and remove the unrelated MySQL-side behavior change, I’d be happy to re-review the next head.

[PZh101]

Thank you for your review. Regarding question one, I have a few points to make. Because when I tested on the Kingbase 8 version, I found that if the current fetchsize is set to Integer.MIN_VALUE, the driver of Kingbase will report an error. So I added this piece of code. In the future, I will create a new class related to the MySQL compatibility mode for Kingbase based on the current MysqlDialect to handle it, and remove the judgment in MysqlDialect.

[DanielLeens]

Thanks for the clarification, and the extra Kingbase driver detail is helpful. If Integer.MIN_VALUE really breaks the Kingbase 8 driver in MySQL-compatible mode, then the underlying runtime problem is absolutely real. My concern is narrower: on the current head, the fix is still implemented by changing the live MysqlDialect.creatPreparedStatement(...) path, so the PR now alters real MySQL source behavior before a Kingbase-specific branch exists. That is why I still need a code change here before I can clear the blocker. If you want to keep the current scope, please move this fetch-size handling into the Kingbase-specific runtime path (or introduce the separate Kingbase MySQL-compatible dialect you mentioned) in the PR itself, then I am happy to re-review the next head. Until there is a new commit, I would keep the current review conclusion unchanged.

[PZh101]

Regarding the compatibility issues identified during the local testing, I have reverted the modifications to MySQLDialect and adjusted KingbaseCatalog. To ensure the independence and cohesion of the code, all relevant changes have now been consolidated within the KingBase class, without affecting other dialect modules.
In the MySQL compatibility mode, the system can automatically and dynamically obtain the compatibleMode through the current connection. The changes have been pushed. Please take the time to review this PR again when you have the chance. Thank you.

[DanielLeens]

Thanks for the follow-up and for narrowing the Kingbase-specific runtime path. I re-reviewed the latest head from the real JDBC source/sink call chain again.

What this PR solves

• User pain: KingbaseES running in MySQL compatibility mode should behave like the MySQL-compatible JDBC path instead of silently falling back to default Kingbase behavior.
• Fix approach: detect compatibility mode from the real Kingbase connection when needed and construct an immutable Kingbase dialect with that result.
• One-line summary: the Kingbase wiring is much better now, but one unrelated MySQL runtime change is still in the PR, and the user-facing docs are still incomplete.

Runtime path I checked

Kingbase source/sink/catalog init
  -> JdbcDialectLoader.load(..., jdbcConnectionConfig)
      -> KingbaseDialectFactory.create(...)
          -> detect compatible mode
          -> new KingbaseDialect(...)

separate MySQL source path still touched by this PR
  -> JdbcInputFormat.open(...)
      -> ChunkSplitter.createPreparedStatement(...)
          -> jdbcDialect.creatPreparedStatement(...)
              -> MysqlDialect.creatPreparedStatement(...)

Findings

Problem 1: the PR still changes the real MySQL source fetch path even though that behavior is outside the Kingbase feature boundary

• Location: seatunnel-connectors-v2/connector-jdbc/.../mysql/MysqlDialect.java:123-130
• Severity: High
• Raised before by others: No
• Why this is still a blocker:
  the current head still changes MysqlDialect.creatPreparedStatement(...) from unconditional Integer.MIN_VALUE streaming mode to setFetchSize(fetchSize) whenever fetch_size > 0. That is a live MySQL runtime behavior change, and it still does not belong to the Kingbase-specific compatibility fix itself.
• Better fix:
  keep the Kingbase runtime wiring, but revert this MySQL-side behavior change from the PR. If Kingbase needs a different fetch strategy, keep it in the Kingbase-specific path.

Problem 2: the user-facing docs are still incomplete for a new compatibility feature

• Location:
  • updated: docs/en/connectors/source/Kingbase.md, docs/zh/connectors/source/Kingbase.md
  • still missing: docs/en/connectors/sink/Kingbase.md, docs/zh/connectors/sink/Kingbase.md
• Severity: Medium
• Raised before by others: No
• Why this matters:
  this PR affects both source and sink runtime wiring, but only the source docs were updated. Users still do not get a complete English/Chinese description of when to set compatible_mode=mysql and what auto-detection does across both paths.
• Better fix:
  complete the sink-side docs as well, and make the source/sink wording consistent.

CI

• The current GitHub Build is red.
• In the metadata available here, I could confirm the red status but not retrieve enough detailed fork-run log text to honestly attribute the current red directly to the Kingbase code path.

Conclusion: can merge after fixes

1. Blocking items

• Problem 1

2. Suggested non-blocking follow-up

• Problem 2

[PZh101]

Thanks for the detailed review!
I believe this issue has been addressed in my latest commit. I've reverted the changes to MysqlDialect.java to keep the original MySQL runtime behavior intact. The Kingbase-specific fetch strategy is now fully isolated within its own dialect path.
Could you please take another look when you have a moment? Thanks!
The image below shows the comparison with my previous edits, and there are no differences anymore.

The image below shows the diff on GitHub. I have already reverted the changes related to fetchSize in MysqlDialect.

[DanielLeens]

Thanks for the detailed follow-up.

I took another look at the current PR state, and the key point here is that the head is still 76c7c8524b86444261d75d2100209d0f0094b62c, which is the same revision I already reviewed in the latest Daniel comment. So from my side this is still a reply-only follow-up on the unchanged head, not a fresh full re-review.

If the MySQL-side change has really been removed from the effective PR code now, please push the follow-up commit (or point to the exact new commit once it lands) and I will gladly re-review that new head from scratch. On the current unchanged revision, the existing Daniel blocker remains the same: I still need the Kingbase-specific fix to stay isolated from the live MysqlDialect runtime path.

[PZh101]

Thanks for the review!
I've double-checked the commits, and it appears that the adjustments to MysqlDialect were actually included in commit [80183ad9f5293e4fc53e8d240663635219413c1d ]. It's possible that the previous view might not have refreshed correctly.
I've attached a screenshot from my repository below as proof. Could you please take another look? Thanks!

[DanielLeens]

Thanks for double-checking this. I rechecked the commit list on the current PR branch before replying. Commit 80183ad9f5293e4fc53e8d240663635219413c1d is indeed part of this PR history, but the current head is still 76c7c8524b86444261d75d2100209d0f0094b62c, which is the same later revision I already reviewed on May 31. So this was not a stale-view issue on Daniel's side: the latest Daniel review already covered a head that includes 80183ad9....

Because there still has not been a new commit after that latest Daniel review, I do not want to pretend a fresh full re-review happened on the same code version. The current Daniel blocker therefore remains the same as in the latest review: the PR still changes the real MySQL runtime fetch path in MysqlDialect.creatPreparedStatement(...), and that behavior change still sits outside the Kingbase-only feature boundary.

If you push a new commit that keeps the Kingbase-specific compatibility path but removes the unrelated MySQL runtime change, I will happily re-review the latest head again.

[PZh101]

I have cleaned up the PR history and pushed a new head.

• Squashed the previous commits into one focused commit.
• MysqlDialect.java is no longer touched by this PR.
• The Kingbase MySQL compatibility logic is kept in the Kingbase path.
• Added both source and sink documentation in English and Chinese.
• Verified with KingbaseDialectTest and KingbaseDialectFactoryTest.

New head: e9f509651ca662d9798b10678b508a693c4b1151

Thanks for the review.

[DanielLeens]

Thanks for the update. I re-reviewed the latest head from scratch and checked the full current source / sink / catalog path again, especially the earlier concern that the change leaked into the real MySQL runtime path.

What this PR solves

• User pain: KingbaseES in MySQL compatibility mode can expose MySQL-like SQL behavior, but SeaTunnel still treated it as plain Kingbase in the JDBC source/sink/catalog path.
• Fix approach: keep the compatibility logic inside the Kingbase path, detect the compatible level from the Kingbase connection when needed, and switch the row converter / type mapper / upsert SQL / catalog only for that Kingbase mode.
• One-line summary: the latest head fixes my earlier “this changes the real MySQL path” blocker; I do not see a new source-level blocker, but the branch still needs a clean CI signal.

Runtime path I checked

source side
  -> JdbcInputFormat(...) [70-75]
      -> JdbcDialectLoader.load(..., jdbcConnectionConfig)

sink side
  -> JdbcSinkFactory.createSink() [185-191]
      -> JdbcDialectLoader.load(..., sinkConfig.getJdbcConnectionConfig())

Kingbase dialect creation
  -> KingbaseDialectFactory.create(..., jdbcConnectionConfig) [62-69]
      -> detectCompatibleMode(...) [71-87]
      -> KbConnection.getCompatibleLevel()

runtime behavior switch
  -> KingbaseDialect.getRowConverter() [57-62]
  -> KingbaseDialect.getJdbcDialectTypeMapper() [65-70]
  -> KingbaseDialect.getUpsertStatement() [73-95]

catalog path
  -> KingbaseCatalogFactory.createCatalog() [52-69]
      -> MySqlCatalog only when compatible level is mysql

Re-review result

• The latest head no longer touches MysqlDialect.java, so the old boundary problem is gone.
• The MySQL-compatible behavior is now scoped inside the Kingbase dialect and catalog path:
  • KingbaseDialectFactory.java:62-69 builds the dialect with the detected compatible level.
  • KingbaseDialect.java:57-76 switches the row converter / type mapper / upsert SQL only for that Kingbase mode.
  • KingbaseCatalogFactory.java:58-69 only returns MySqlCatalog for the Kingbase MySQL-compatible case.
• The new source and sink docs are present in both English and Chinese.
• I did not find a reopened source-level blocker on the latest head.

Tests / CI

• The added unit coverage in KingbaseDialectFactoryTest and KingbaseDialectTest is helpful for the explicit-mode and compatibility-switch paths.
• I did not see a flaky-test pattern in the new unit tests.
• The current fork Build is still red, with failures in unit-test (8, ubuntu-latest) and kafka-connector-it (8, ubuntu-latest), so the branch still needs a clean CI signal before merge.

Conclusion: can merge after fixes

1. Blocking items

• No new source-level blocker from my side on the latest head.
• Please get the current Build green before merge.

2. Suggested non-blocking follow-up

• None from my side on the current code path.

The latest revision is clearly safer than the earlier version. The remaining blocker for me is CI cleanliness, not the Kingbase compatibility design on the latest head.