This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new be5d0f0279 [Feature][Connector-V2] Add Google Cloud Bigtable Source
and Sink con… (#10849)
be5d0f0279 is described below
commit be5d0f0279d72ba6f6b64e22f44bfc4391117816
Author: 雷炯 <[email protected]>
AuthorDate: Sat Jun 20 13:04:10 2026 +0800
[Feature][Connector-V2] Add Google Cloud Bigtable Source and Sink con…
(#10849)
Co-authored-by: leijiong <[email protected]>
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
Co-authored-by: Cursor <[email protected]>
---
.dlc.json | 3 +-
config/plugin_config | 1 +
.../changelog/connector-google-bigtable.md | 7 +
docs/en/connectors/sink/GoogleBigtable.md | 176 ++++++++++++
docs/en/connectors/source/GoogleBigtable.md | 139 ++++++++++
.../changelog/connector-google-bigtable.md | 7 +
docs/zh/connectors/sink/GoogleBigtable.md | 105 ++++++++
docs/zh/connectors/source/GoogleBigtable.md | 59 ++++
plugin-mapping.properties | 2 +
.../connector-google-bigtable/pom.xml | 64 +++++
.../seatunnel/bigtable/client/BigtableClient.java | 173 ++++++++++++
.../bigtable/config/BigtableBaseOptions.java | 61 +++++
.../bigtable/config/BigtableParameters.java | 94 +++++++
.../bigtable/config/BigtableSinkOptions.java | 90 +++++++
.../bigtable/config/BigtableSourceOptions.java | 64 +++++
.../bigtable/constant/BigtableIdentifier.java | 24 ++
.../exception/BigtableConnectorErrorCode.java | 51 ++++
.../exception/BigtableConnectorException.java | 32 +++
.../format/BigtableDeserializationFormat.java | 119 ++++++++
.../seatunnel/bigtable/sink/BigtableSink.java | 127 +++++++++
.../bigtable/sink/BigtableSinkFactory.java | 67 +++++
.../bigtable/sink/BigtableSinkWriter.java | 298 +++++++++++++++++++++
.../seatunnel/bigtable/source/BigtableSource.java | 81 ++++++
.../bigtable/source/BigtableSourceFactory.java | 78 ++++++
.../bigtable/source/BigtableSourceReader.java | 254 ++++++++++++++++++
.../bigtable/source/BigtableSourceSplit.java | 60 +++++
.../source/BigtableSourceSplitEnumerator.java | 176 ++++++++++++
.../bigtable/source/BigtableSourceState.java | 36 +++
.../state/BigtableAggregatedCommitInfo.java | 24 ++
.../bigtable/state/BigtableCommitInfo.java | 24 ++
.../bigtable/state/BigtableSinkState.java | 24 ++
.../bigtable/sink/BigtableSinkSaveModeTest.java | 153 +++++++++++
.../bigtable/sink/BigtableSinkWriterTest.java | 129 +++++++++
.../bigtable/source/BigtableSourceReaderTest.java | 267 ++++++++++++++++++
.../source/BigtableSourceSplitEnumeratorTest.java | 40 +++
seatunnel-connectors-v2/pom.xml | 1 +
seatunnel-dist/pom.xml | 6 +
37 files changed, 3115 insertions(+), 1 deletion(-)
diff --git a/.dlc.json b/.dlc.json
index b25f2512f2..a43af66641 100644
--- a/.dlc.json
+++ b/.dlc.json
@@ -42,6 +42,7 @@
0,
200,
401,
- 403
+ 403,
+ 406
]
}
diff --git a/config/plugin_config b/config/plugin_config
index fc6bf5c7c2..5f560a4a46 100644
--- a/config/plugin_config
+++ b/config/plugin_config
@@ -47,6 +47,7 @@ connector-file-sftp
connector-file-obs
connector-google-sheets
connector-google-firestore
+connector-google-bigtable
connector-graphql
connector-hive
connector-http-base
diff --git a/docs/en/connectors/changelog/connector-google-bigtable.md
b/docs/en/connectors/changelog/connector-google-bigtable.md
new file mode 100644
index 0000000000..3fc3a897f9
--- /dev/null
+++ b/docs/en/connectors/changelog/connector-google-bigtable.md
@@ -0,0 +1,7 @@
+<details><summary> Change Log </summary>
+
+| Change | Commit | Version |
+| --- | --- | --- |
+|[Feature][Connector-V2] Add Google Cloud Bigtable Source and Sink
connector|https://github.com/apache/seatunnel/commit/8e57c04|dev|
+
+</details>
diff --git a/docs/en/connectors/sink/GoogleBigtable.md
b/docs/en/connectors/sink/GoogleBigtable.md
new file mode 100644
index 0000000000..55b96cf398
--- /dev/null
+++ b/docs/en/connectors/sink/GoogleBigtable.md
@@ -0,0 +1,176 @@
+import ChangeLog from '../changelog/connector-google-bigtable.md';
+
+# GoogleBigtable
+
+> Google Bigtable sink connector
+
+## Description
+
+Writes data to Google Cloud Bigtable using the native Bigtable Data v2 Java
client.
+
+## Key features
+
+- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [x] [batch](../../introduction/concepts/connector-v2-features.md)
+
+## Options
+
+| name | type | required | default value |
+|---------------------|---------|----------|---------------|
+| project_id | string | yes | - |
+| instance_id | string | yes | - |
+| table | string | yes | - |
+| rowkey_column | list | yes | - |
+| column_family | config | yes | - |
+| credentials_path | string | no | - |
+| rowkey_delimiter | string | no | "" |
+| version_column | string | no | - |
+| null_mode | string | no | skip |
+| batch_mutation_size | int | no | 100 |
+| common-options | | no | - |
+
+### project_id [string]
+
+Google Cloud project ID. Example: `"my-gcp-project"`
+
+### instance_id [string]
+
+Bigtable instance ID. Example: `"my-bigtable-instance"`
+
+### table [string]
+
+The Bigtable table name to write to. Example: `"my-table"`
+
+### rowkey_column [list]
+
+Column names used to compose the Bigtable row key. Example: `["id"]` or
`["tenant", "id"]`.
+
+When multiple columns are specified they are joined with `rowkey_delimiter`.
+
+### column_family [config]
+
+Mapping from column name to column family name. Use `all_columns` as key to
set a default family for all unmapped columns.
+
+```hocon
+column_family {
+ name = "info"
+ age = "stats"
+}
+```
+
+or to put everything in one family:
+
+```hocon
+column_family {
+ all_columns = "cf"
+}
+```
+
+### credentials_path [string]
+
+Path to the Google Cloud service account JSON key file.
+
+If not set, [Application Default Credentials
(ADC)](https://cloud.google.com/docs/authentication/application-default-credentials)
will be used — this works automatically on GCE/GKE or when
`GOOGLE_APPLICATION_CREDENTIALS` is set in the environment.
+
+### rowkey_delimiter [string]
+
+Delimiter used to join multiple row-key column values. Default is `""` (empty
string, no delimiter).
+
+### version_column [string]
+
+Column name whose `BIGINT` value is used as the Bigtable cell timestamp
(microseconds since epoch). If not set, the current system time is used.
+
+### null_mode [string]
+
+How to handle `null` field values. Supported: `skip` (default), `empty`.
+
+- `skip` — the cell is omitted from the mutation
+- `empty` — an empty byte array is written to the cell
+
+### batch_mutation_size [int]
+
+Number of row mutations to accumulate before sending a BulkMutation to
Bigtable. Default is `100`. Increase for higher throughput at the cost of
higher per-task memory usage.
+
+### common options
+
+Sink plugin common parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details.
+
+## Data Types
+
+All SeaTunnel types are supported:
+
+| SeaTunnel type | Storage format in Bigtable |
+|------------------------------|---------------------------------|
+| TINYINT | 1-byte binary |
+| SMALLINT | 2-byte big-endian binary |
+| INT | 4-byte big-endian binary |
+| BIGINT | 8-byte big-endian binary |
+| FLOAT | 4-byte IEEE 754 big-endian |
+| DOUBLE | 8-byte IEEE 754 big-endian |
+| BOOLEAN | 1-byte (1 = true, 0 = false) |
+| BYTES | Raw bytes |
+| STRING | UTF-8 text |
+| DECIMAL | UTF-8 plain string |
+| DATE | UTF-8 `yyyy-MM-dd` |
+| TIME | UTF-8 `HH:mm:ss` |
+| TIMESTAMP | UTF-8 `yyyy-MM-dd HH:mm:ss` |
+
+## Example
+
+### Basic — Application Default Credentials
+
+```hocon
+sink {
+ GoogleBigtable {
+ project_id = "my-gcp-project"
+ instance_id = "my-bigtable-instance"
+ table = "events"
+ rowkey_column = ["event_id"]
+ column_family {
+ all_columns = "cf"
+ }
+ }
+}
+```
+
+### Service Account Key File
+
+```hocon
+sink {
+ GoogleBigtable {
+ project_id = "my-gcp-project"
+ instance_id = "my-bigtable-instance"
+ table = "events"
+ credentials_path = "/secrets/sa-key.json"
+ rowkey_column = ["tenant_id", "event_id"]
+ rowkey_delimiter = "#"
+ column_family {
+ all_columns = "data"
+ }
+ batch_mutation_size = 500
+ }
+}
+```
+
+### Multiple Column Families
+
+```hocon
+sink {
+ GoogleBigtable {
+ project_id = "my-gcp-project"
+ instance_id = "my-bigtable-instance"
+ table = "user_profile"
+ rowkey_column = ["user_id"]
+ column_family {
+ name = "identity"
+ email = "identity"
+ age = "stats"
+ last_login = "stats"
+ }
+ }
+}
+```
+
+## Changelog
+
+<ChangeLog />
diff --git a/docs/en/connectors/source/GoogleBigtable.md
b/docs/en/connectors/source/GoogleBigtable.md
new file mode 100644
index 0000000000..b511aa38c7
--- /dev/null
+++ b/docs/en/connectors/source/GoogleBigtable.md
@@ -0,0 +1,139 @@
+import ChangeLog from '../changelog/connector-google-bigtable.md';
+
+# GoogleBigtable
+
+> Google Bigtable source connector
+
+## Description
+
+Reads data from Google Cloud Bigtable using the native Bigtable Data v2 Java
client.
+
+## Key features
+
+- [x] [batch](../../introduction/concepts/connector-v2-features.md)
+- [ ] [stream](../../introduction/concepts/connector-v2-features.md)
+- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [ ] [parallelism](../../introduction/concepts/connector-v2-features.md)
+
+## Options
+
+| name | type | required | default value |
+|------------------|--------|----------|---------------|
+| project_id | string | yes | - |
+| instance_id | string | yes | - |
+| table | string | yes | - |
+| credentials_path | string | no | - |
+| rowkey_column | list | no | - |
+| start_rowkey | string | no | - |
+| end_rowkey | string | no | - |
+| start_timestamp | long | no | - |
+| end_timestamp | long | no | - |
+| max_versions | int | no | 1 |
+| scan_row_limit | int | no | -1 |
+| common-options | | no | - |
+
+### project_id [string]
+
+Google Cloud project ID.
+
+### instance_id [string]
+
+Bigtable instance ID.
+
+### table [string]
+
+Bigtable table name to read from.
+
+### credentials_path [string]
+
+Path to the Google Cloud service account JSON key file. If omitted,
Application Default Credentials (ADC) are used.
+
+### rowkey_column [list]
+
+Optional list of field names that should receive the row key value. Declare a
field named `rowkey` in your schema to capture the raw row key bytes as a
`BYTES` or `STRING` field.
+
+### start_rowkey [string]
+
+Inclusive start row key for the scan. If not set, the scan starts from the
beginning of the table.
+
+### end_rowkey [string]
+
+Exclusive end row key for the scan. If not set, the scan reads to the end of
the table.
+
+### start_timestamp [long]
+
+Inclusive start timestamp filter (microseconds since epoch).
+
+### end_timestamp [long]
+
+Exclusive end timestamp filter (microseconds since epoch).
+
+### max_versions [int]
+
+Maximum number of cell versions to return per column qualifier. Default `1`
returns only the latest version.
+
+### scan_row_limit [int]
+
+Maximum number of rows to return. `-1` (default) means no limit.
+
+### common options
+
+Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
+
+## Schema Mapping
+
+Field names in the SeaTunnel schema must follow the pattern
`familyName:qualifier`, for example `cf:name` or `stats:age`. The special field
name `rowkey` maps to the Bigtable row key.
+
+| Schema field name | Mapped Bigtable cell |
+|-------------------|-----------------------------|
+| `rowkey` | Row key |
+| `cf:name` | Column family `cf`, qualifier `name` |
+| `stats:age` | Column family `stats`, qualifier `age` |
+
+## Example
+
+### Read all rows — Application Default Credentials
+
+```hocon
+source {
+ GoogleBigtable {
+ project_id = "my-gcp-project"
+ instance_id = "my-bigtable-instance"
+ table = "events"
+ schema {
+ fields {
+ rowkey = BYTES
+ "cf:type" = STRING
+ "cf:ts" = BIGINT
+ }
+ }
+ }
+}
+```
+
+### Scan a row-key range with a service account
+
+```hocon
+source {
+ GoogleBigtable {
+ project_id = "my-gcp-project"
+ instance_id = "my-bigtable-instance"
+ table = "events"
+ credentials_path = "/secrets/sa-key.json"
+ start_rowkey = "2024-01-01#"
+ end_rowkey = "2024-02-01#"
+ max_versions = 1
+ schema {
+ fields {
+ rowkey = STRING
+ "cf:type" = STRING
+ "cf:data" = STRING
+ }
+ }
+ }
+}
+```
+
+## Changelog
+
+<ChangeLog />
diff --git a/docs/zh/connectors/changelog/connector-google-bigtable.md
b/docs/zh/connectors/changelog/connector-google-bigtable.md
new file mode 100644
index 0000000000..3fc3a897f9
--- /dev/null
+++ b/docs/zh/connectors/changelog/connector-google-bigtable.md
@@ -0,0 +1,7 @@
+<details><summary> Change Log </summary>
+
+| Change | Commit | Version |
+| --- | --- | --- |
+|[Feature][Connector-V2] Add Google Cloud Bigtable Source and Sink
connector|https://github.com/apache/seatunnel/commit/8e57c04|dev|
+
+</details>
diff --git a/docs/zh/connectors/sink/GoogleBigtable.md
b/docs/zh/connectors/sink/GoogleBigtable.md
new file mode 100644
index 0000000000..65fd4f927d
--- /dev/null
+++ b/docs/zh/connectors/sink/GoogleBigtable.md
@@ -0,0 +1,105 @@
+import ChangeLog from '../changelog/connector-google-bigtable.md';
+
+# GoogleBigtable
+
+> Google Bigtable Sink 连接器
+
+## 描述
+
+使用原生 Bigtable Data v2 Java 客户端将数据写入 Google Cloud Bigtable。
+
+## 主要特性
+
+- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [x] [batch](../../introduction/concepts/connector-v2-features.md)
+
+## 参数
+
+| 参数名 | 类型 | 是否必填 | 默认值 |
+|--------------------|---------|--------|------|
+| project_id | string | 是 | - |
+| instance_id | string | 是 | - |
+| table | string | 是 | - |
+| rowkey_column | list | 是 | - |
+| column_family | config | 是 | - |
+| credentials_path | string | 否 | - |
+| rowkey_delimiter | string | 否 | "" |
+| version_column | string | 否 | - |
+| null_mode | string | 否 | skip |
+| batch_mutation_size| int | 否 | 100 |
+| common-options | | 否 | - |
+
+### project_id [string]
+
+Google Cloud 项目 ID，例如 `"my-gcp-project"`。
+
+### instance_id [string]
+
+Bigtable 实例 ID，例如 `"my-bigtable-instance"`。
+
+### table [string]
+
+写入的 Bigtable 表名，例如 `"my-table"`。
+
+### rowkey_column [list]
+
+用于构造行键的列名列表，例如 `["id"]` 或 `["tenant_id", "event_id"]`。多列时用 `rowkey_delimiter`
拼接。
+
+### column_family [config]
+
+列名到列族的映射配置。可使用 `all_columns` 作为默认列族：
+
+```hocon
+column_family {
+ all_columns = "cf"
+}
+```
+
+也可以为不同列指定不同列族：
+
+```hocon
+column_family {
+ name = "info"
+ age = "stats"
+}
+```
+
+### credentials_path [string]
+
+Google Cloud 服务账号 JSON 密钥文件路径。未设置时使用应用默认凭证（ADC）。
+
+### rowkey_delimiter [string]
+
+多列行键的拼接分隔符，默认为空字符串 `""`。
+
+### version_column [string]
+
+用作 Bigtable Cell 时间戳（微秒）的 BIGINT 列名。未设置时使用当前系统时间。
+
+### null_mode [string]
+
+空值写入策略：`skip`（默认，跳过该 Cell）或 `empty`（写入空字节数组）。
+
+### batch_mutation_size [int]
+
+每次批量提交的行数，默认 `100`。
+
+## 示例
+
+```hocon
+sink {
+ GoogleBigtable {
+ project_id = "my-gcp-project"
+ instance_id = "my-bigtable-instance"
+ table = "events"
+ rowkey_column = ["event_id"]
+ column_family {
+ all_columns = "cf"
+ }
+ }
+}
+```
+
+## Changelog
+
+<ChangeLog />
diff --git a/docs/zh/connectors/source/GoogleBigtable.md
b/docs/zh/connectors/source/GoogleBigtable.md
new file mode 100644
index 0000000000..ea1b4e8190
--- /dev/null
+++ b/docs/zh/connectors/source/GoogleBigtable.md
@@ -0,0 +1,59 @@
+import ChangeLog from '../changelog/connector-google-bigtable.md';
+
+# GoogleBigtable
+
+> Google Bigtable Source 连接器
+
+## 描述
+
+使用原生 Bigtable Data v2 Java 客户端从 Google Cloud Bigtable 读取数据。
+
+## 主要特性
+
+- [x] [batch](../../introduction/concepts/connector-v2-features.md)
+- [ ] [parallelism](../../introduction/concepts/connector-v2-features.md)
+- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+
+## 参数
+
+| 参数名 | 类型 | 是否必填 | 默认值 |
+|-----------------|--------|--------|------|
+| project_id | string | 是 | - |
+| instance_id | string | 是 | - |
+| table | string | 是 | - |
+| credentials_path| string | 否 | - |
+| rowkey_column | list | 否 | - |
+| start_rowkey | string | 否 | - |
+| end_rowkey | string | 否 | - |
+| start_timestamp | long | 否 | - |
+| end_timestamp | long | 否 | - |
+| max_versions | int | 否 | 1 |
+| scan_row_limit | int | 否 | -1 |
+| common-options | | 否 | - |
+
+### Schema 映射
+
+字段名须使用 `列族:列限定符` 格式，例如 `cf:name`、`stats:age`。特殊字段名 `rowkey` 映射到行键。
+
+## 示例
+
+```hocon
+source {
+ GoogleBigtable {
+ project_id = "my-gcp-project"
+ instance_id = "my-bigtable-instance"
+ table = "events"
+ schema {
+ fields {
+ rowkey = BYTES
+ "cf:type" = STRING
+ "cf:ts" = BIGINT
+ }
+ }
+ }
+}
+```
+
+## Changelog
+
+<ChangeLog />
diff --git a/plugin-mapping.properties b/plugin-mapping.properties
index 800f50b7be..5e0a4460ce 100644
--- a/plugin-mapping.properties
+++ b/plugin-mapping.properties
@@ -90,6 +90,8 @@ seatunnel.source.MyHours = connector-http-myhours
seatunnel.sink.InfluxDB = connector-influxdb
seatunnel.source.GoogleSheets = connector-google-sheets
seatunnel.sink.GoogleFirestore = connector-google-firestore
+seatunnel.source.GoogleBigtable = connector-google-bigtable
+seatunnel.sink.GoogleBigtable = connector-google-bigtable
seatunnel.sink.Tablestore = connector-tablestore
seatunnel.source.Tablestore = connector-tablestore
seatunnel.source.Lemlist = connector-http-lemlist
diff --git a/seatunnel-connectors-v2/connector-google-bigtable/pom.xml
b/seatunnel-connectors-v2/connector-google-bigtable/pom.xml
new file mode 100644
index 0000000000..5f6e03aaaa
--- /dev/null
+++ b/seatunnel-connectors-v2/connector-google-bigtable/pom.xml
@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <artifactId>connector-google-bigtable</artifactId>
+ <name>SeaTunnel : Connectors V2 : Google Bigtable</name>
+
+ <properties>
+ <bigtable.version>2.39.1</bigtable.version>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.cloud</groupId>
+ <artifactId>google-cloud-bigtable</artifactId>
+ <version>${bigtable.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+</project>
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/client/BigtableClient.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/client/BigtableClient.java
new file mode 100644
index 0000000000..d4d21d9718
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/client/BigtableClient.java
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.client;
+
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorException;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.ServiceAccountCredentials;
+import com.google.cloud.bigtable.data.v2.BigtableDataClient;
+import com.google.cloud.bigtable.data.v2.BigtableDataSettings;
+import com.google.cloud.bigtable.data.v2.models.BulkMutation;
+import com.google.cloud.bigtable.data.v2.models.Mutation;
+import com.google.cloud.bigtable.data.v2.models.RowMutation;
+import com.google.protobuf.ByteString;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * Wrapper around the native Google Cloud Bigtable Java client.
+ *
+ * <p>Provides basic CRUD operations and batched mutation support used by both
the Sink and Source
+ * connectors.
+ */
+@Slf4j
+public class BigtableClient implements Serializable, AutoCloseable {
+
+ private transient BigtableDataClient dataClient;
+ private final BigtableParameters parameters;
+
+ private BigtableClient(BigtableDataClient dataClient, BigtableParameters
parameters) {
+ this.dataClient = dataClient;
+ this.parameters = parameters;
+ }
+
+ /**
+ * Creates a new BigtableClient instance using the provided parameters.
+ *
+ * @param parameters Bigtable connection parameters
+ * @return a connected BigtableClient
+ */
+ public static BigtableClient createInstance(BigtableParameters parameters)
{
+ try {
+ BigtableDataSettings.Builder settingsBuilder =
+ BigtableDataSettings.newBuilder()
+ .setProjectId(parameters.getProjectId())
+ .setInstanceId(parameters.getInstanceId());
+
+ if (parameters.getCredentialsPath() != null
+ && !parameters.getCredentialsPath().isEmpty()) {
+ try (FileInputStream credStream =
+ new FileInputStream(parameters.getCredentialsPath())) {
+ GoogleCredentials credentials =
+ ServiceAccountCredentials.fromStream(credStream);
+ settingsBuilder.stubSettings().setCredentialsProvider(()
-> credentials);
+ }
+ }
+
+ BigtableDataClient client =
BigtableDataClient.create(settingsBuilder.build());
+ return new BigtableClient(client, parameters);
+ } catch (IOException e) {
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.CONNECTION_FAILED,
+ "Failed to create Bigtable client for project="
+ + parameters.getProjectId()
+ + ", instance="
+ + parameters.getInstanceId(),
+ e);
+ }
+ }
+
+ /**
+ * Applies a single row mutation to Bigtable.
+ *
+ * @param rowMutation the row mutation to apply
+ */
+ public void mutateRow(RowMutation rowMutation) {
+ try {
+ dataClient.mutateRow(rowMutation);
+ } catch (Exception e) {
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.WRITE_FAILED,
+ "Failed to mutate row in table " + parameters.getTable(),
+ e);
+ }
+ }
+
+ /**
+ * Applies a batch of row mutations to Bigtable using BulkMutation for
efficiency.
+ *
+ * @param mutations list of (rowKey, Mutation) pairs to apply
+ */
+ public void bulkMutate(List<RowKeyMutation> mutations) {
+ if (mutations.isEmpty()) {
+ return;
+ }
+ try {
+ BulkMutation bulk = BulkMutation.create(parameters.getTable());
+ for (RowKeyMutation entry : mutations) {
+ bulk.add(entry.getRowKey(), entry.getMutation());
+ }
+ dataClient.bulkMutateRows(bulk);
+ } catch (Exception e) {
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.WRITE_FAILED,
+ "Failed to bulk mutate "
+ + mutations.size()
+ + " rows in table "
+ + parameters.getTable(),
+ e);
+ }
+ }
+
+ /**
+ * Returns the underlying data client (for use in Source reader streaming).
+ *
+ * @return BigtableDataClient
+ */
+ public BigtableDataClient getDataClient() {
+ return dataClient;
+ }
+
+ @Override
+ public void close() {
+ if (dataClient != null) {
+ try {
+ dataClient.close();
+ dataClient = null;
+ } catch (Exception e) {
+ log.error("Failed to close Bigtable data client", e);
+ }
+ }
+ }
+
+ /** Holds a row key and its associated Mutation for bulk operations. */
+ public static class RowKeyMutation implements Serializable {
+ private final ByteString rowKey;
+ private final Mutation mutation;
+
+ public RowKeyMutation(ByteString rowKey, Mutation mutation) {
+ this.rowKey = rowKey;
+ this.mutation = mutation;
+ }
+
+ public ByteString getRowKey() {
+ return rowKey;
+ }
+
+ public Mutation getMutation() {
+ return mutation;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableBaseOptions.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableBaseOptions.java
new file mode 100644
index 0000000000..01850ee698
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableBaseOptions.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.options.ConnectorCommonOptions;
+
+import java.util.List;
+
+public class BigtableBaseOptions extends ConnectorCommonOptions {
+
+ public static final Option<String> PROJECT_ID =
+ Options.key("project_id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Google Cloud project ID");
+
+ public static final Option<String> INSTANCE_ID =
+ Options.key("instance_id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Bigtable instance ID");
+
+ public static final Option<String> TABLE =
+ Options.key("table")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Bigtable table name");
+
+ public static final Option<List<String>> ROWKEY_COLUMNS =
+ Options.key("rowkey_column")
+ .listType()
+ .noDefaultValue()
+ .withDescription(
+ "Column names used to compose the Bigtable row
key. "
+ + "If multiple columns are specified they
are joined with rowkey_delimiter.");
+
+ public static final Option<String> CREDENTIALS_PATH =
+ Options.key("credentials_path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Path to the Google Cloud service account JSON key
file. "
+ + "If not set, Application Default
Credentials (ADC) will be used.");
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableParameters.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableParameters.java
new file mode 100644
index 0000000000..69a005976a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableParameters.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.config;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+
+import lombok.Builder;
+import lombok.Getter;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+@Builder
+@Getter
+public class BigtableParameters implements Serializable {
+
+ private String projectId;
+ private String instanceId;
+ private String table;
+ private List<String> rowkeyColumns;
+ private String credentialsPath;
+
+ // Sink-specific
+ private Map<String, String> columnFamily;
+ @Builder.Default private String rowkeyDelimiter = "";
+ private String versionColumn;
+
+ @Builder.Default
+ private BigtableSinkOptions.NullMode nullMode =
BigtableSinkOptions.NullMode.SKIP;
+
+ @Builder.Default private int batchMutationSize = 100;
+
+ // Source-specific
+ private String startRowkey;
+ private String endRowkey;
+ private Long startTimestamp;
+ private Long endTimestamp;
+ @Builder.Default private int maxVersions = 1;
+ @Builder.Default private int scanRowLimit = -1;
+
+ public static BigtableParameters buildWithConfig(ReadonlyConfig config) {
+ BigtableParametersBuilder builder = BigtableParameters.builder();
+ builder.projectId(config.get(BigtableBaseOptions.PROJECT_ID));
+ builder.instanceId(config.get(BigtableBaseOptions.INSTANCE_ID));
+ builder.table(config.get(BigtableBaseOptions.TABLE));
+ builder.rowkeyColumns(config.get(BigtableBaseOptions.ROWKEY_COLUMNS));
+
+ config.getOptional(BigtableBaseOptions.CREDENTIALS_PATH)
+ .ifPresent(builder::credentialsPath);
+
config.getOptional(BigtableSinkOptions.COLUMN_FAMILY).ifPresent(builder::columnFamily);
+ config.getOptional(BigtableSinkOptions.ROWKEY_DELIMITER)
+ .ifPresent(builder::rowkeyDelimiter);
+
config.getOptional(BigtableSinkOptions.VERSION_COLUMN).ifPresent(builder::versionColumn);
+
config.getOptional(BigtableSinkOptions.NULL_MODE).ifPresent(builder::nullMode);
+ config.getOptional(BigtableSinkOptions.BATCH_MUTATION_SIZE)
+ .ifPresent(builder::batchMutationSize);
+ return builder.build();
+ }
+
+ public static BigtableParameters buildWithSourceConfig(ReadonlyConfig
config) {
+ BigtableParametersBuilder builder = BigtableParameters.builder();
+ builder.projectId(config.get(BigtableBaseOptions.PROJECT_ID));
+ builder.instanceId(config.get(BigtableBaseOptions.INSTANCE_ID));
+ builder.table(config.get(BigtableBaseOptions.TABLE));
+
+
config.getOptional(BigtableBaseOptions.ROWKEY_COLUMNS).ifPresent(builder::rowkeyColumns);
+ config.getOptional(BigtableBaseOptions.CREDENTIALS_PATH)
+ .ifPresent(builder::credentialsPath);
+
config.getOptional(BigtableSourceOptions.START_ROW_KEY).ifPresent(builder::startRowkey);
+
config.getOptional(BigtableSourceOptions.END_ROW_KEY).ifPresent(builder::endRowkey);
+ config.getOptional(BigtableSourceOptions.START_TIMESTAMP)
+ .ifPresent(builder::startTimestamp);
+
config.getOptional(BigtableSourceOptions.END_TIMESTAMP).ifPresent(builder::endTimestamp);
+
config.getOptional(BigtableSourceOptions.MAX_VERSIONS).ifPresent(builder::maxVersions);
+
config.getOptional(BigtableSourceOptions.SCAN_ROW_LIMIT).ifPresent(builder::scanRowLimit);
+ return builder.build();
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableSinkOptions.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableSinkOptions.java
new file mode 100644
index 0000000000..1c3b4bb0b6
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableSinkOptions.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.SchemaSaveMode;
+
+import java.util.Map;
+
+import static org.apache.seatunnel.api.sink.DataSaveMode.APPEND_DATA;
+
+public class BigtableSinkOptions extends BigtableBaseOptions {
+
+ public static final Option<Map<String, String>> COLUMN_FAMILY =
+ Options.key("column_family")
+ .mapType()
+ .noDefaultValue()
+ .withDescription(
+ "Mapping from column name to column family. "
+ + "Use \"all_columns\" as key to set a
default family for unmapped columns.");
+
+ public static final Option<String> ROWKEY_DELIMITER =
+ Options.key("rowkey_delimiter")
+ .stringType()
+ .defaultValue("")
+ .withDescription(
+ "Delimiter used to join multiple rowkey column
values. Default is empty string.");
+
+ public static final Option<String> VERSION_COLUMN =
+ Options.key("version_column")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Column name whose long value is used as the
Bigtable cell timestamp. "
+ + "If not set, the current system time is
used.");
+
+ public static final Option<NullMode> NULL_MODE =
+ Options.key("null_mode")
+ .enumType(NullMode.class)
+ .defaultValue(NullMode.SKIP)
+ .withDescription(
+ "How to handle null field values: SKIP (default)
omits the cell; EMPTY writes an empty byte array.");
+
+ public static final Option<Integer> BATCH_MUTATION_SIZE =
+ Options.key("batch_mutation_size")
+ .intType()
+ .defaultValue(100)
+ .withDescription(
+ "Number of mutations to accumulate before flushing
to Bigtable. Default is 100.");
+
+ public static final Option<SchemaSaveMode> SCHEMA_SAVE_MODE =
+ Options.key("schema_save_mode")
+ .singleChoice(
+ SchemaSaveMode.class,
+
java.util.Arrays.asList(SchemaSaveMode.RECREATE_SCHEMA))
+ .defaultValue(SchemaSaveMode.RECREATE_SCHEMA)
+ .withDescription(
+ "Schema save mode. Only RECREATE_SCHEMA is
currently supported. "
+ + "Table and column families must be
created manually before running the job.");
+
+ public static final Option<DataSaveMode> DATA_SAVE_MODE =
+ Options.key("data_save_mode")
+ .singleChoice(DataSaveMode.class,
java.util.Arrays.asList(APPEND_DATA))
+ .defaultValue(APPEND_DATA)
+ .withDescription(
+ "Data save mode. Only APPEND_DATA is currently
supported. "
+ + "DROP_DATA and ERROR_WHEN_DATA_EXISTS
are not yet implemented.");
+
+ public enum NullMode {
+ SKIP,
+ EMPTY;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableSourceOptions.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableSourceOptions.java
new file mode 100644
index 0000000000..9ef2f88a37
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/config/BigtableSourceOptions.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+public class BigtableSourceOptions extends BigtableBaseOptions {
+
+ public static final Option<String> START_ROW_KEY =
+ Options.key("start_rowkey")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Bigtable scan start row key
(inclusive).");
+
+ public static final Option<String> END_ROW_KEY =
+ Options.key("end_rowkey")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Bigtable scan end row key (exclusive).");
+
+ public static final Option<Long> START_TIMESTAMP =
+ Options.key("start_timestamp")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "Start timestamp (inclusive) for scan time range
in microseconds since epoch.");
+
+ public static final Option<Long> END_TIMESTAMP =
+ Options.key("end_timestamp")
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "End timestamp (exclusive) for scan time range in
microseconds since epoch.");
+
+ public static final Option<Integer> MAX_VERSIONS =
+ Options.key("max_versions")
+ .intType()
+ .defaultValue(1)
+ .withDescription(
+ "Maximum number of cell versions to return per
column. Default is 1 (latest only).");
+
+ public static final Option<Integer> SCAN_ROW_LIMIT =
+ Options.key("scan_row_limit")
+ .intType()
+ .defaultValue(-1)
+ .withDescription(
+ "Maximum number of rows to scan. -1 (default)
means no limit.");
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/constant/BigtableIdentifier.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/constant/BigtableIdentifier.java
new file mode 100644
index 0000000000..6bd287b27a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/constant/BigtableIdentifier.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.constant;
+
+public class BigtableIdentifier {
+ public static final String IDENTIFIER_NAME = "GoogleBigtable";
+
+ private BigtableIdentifier() {}
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/exception/BigtableConnectorErrorCode.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/exception/BigtableConnectorErrorCode.java
new file mode 100644
index 0000000000..0f16864b72
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/exception/BigtableConnectorErrorCode.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+
+public enum BigtableConnectorErrorCode implements SeaTunnelErrorCode {
+ CONNECTION_FAILED("Bigtable-01", "Build Bigtable connection failed"),
+ TABLE_NOT_FOUND("Bigtable-02", "Bigtable table not found"),
+ TABLE_CREATE_FAILED("Bigtable-03", "Bigtable table create failed"),
+ TABLE_DELETE_FAILED("Bigtable-04", "Bigtable table delete failed"),
+ TABLE_TRUNCATE_FAILED("Bigtable-05", "Bigtable table truncate failed"),
+ TABLE_QUERY_FAILED("Bigtable-06", "Bigtable table query failed"),
+ WRITE_FAILED("Bigtable-07", "Bigtable write failed"),
+ READ_FAILED("Bigtable-08", "Bigtable read failed"),
+ CREDENTIALS_FAILED("Bigtable-09", "Failed to load Bigtable credentials"),
+ ;
+
+ private final String code;
+ private final String description;
+
+ BigtableConnectorErrorCode(String code, String description) {
+ this.code = code;
+ this.description = description;
+ }
+
+ @Override
+ public String getCode() {
+ return code;
+ }
+
+ @Override
+ public String getDescription() {
+ return description;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/exception/BigtableConnectorException.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/exception/BigtableConnectorException.java
new file mode 100644
index 0000000000..d9b73e77f3
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/exception/BigtableConnectorException.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+public class BigtableConnectorException extends SeaTunnelRuntimeException {
+
+ public BigtableConnectorException(BigtableConnectorErrorCode errorCode,
String errorMessage) {
+ super(errorCode, errorMessage);
+ }
+
+ public BigtableConnectorException(
+ BigtableConnectorErrorCode errorCode, String errorMessage,
Throwable cause) {
+ super(errorCode, errorMessage, cause);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/format/BigtableDeserializationFormat.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/format/BigtableDeserializationFormat.java
new file mode 100644
index 0000000000..5a7254ba3e
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/format/BigtableDeserializationFormat.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.format;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.DateTimeUtils;
+import org.apache.seatunnel.common.utils.DateUtils;
+import org.apache.seatunnel.common.utils.TimeUtils;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorException;
+
+import com.google.protobuf.ByteString;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * Deserializes raw {@link ByteString} cell values returned by Bigtable into
typed {@link
+ * SeaTunnelRow} fields.
+ *
+ * <p>Numeric types are stored as big-endian binary; string-like types (DATE,
TIME, TIMESTAMP,
+ * DECIMAL, STRING) are stored as UTF-8 text, matching the encoding used by
{@code
+ * BigtableSinkWriter}.
+ */
+public class BigtableDeserializationFormat {
+
+ private final DateUtils.Formatter dateFormat =
DateUtils.Formatter.YYYY_MM_DD;
+ private final DateTimeUtils.Formatter datetimeFormat =
+ DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS;
+ private final TimeUtils.Formatter timeFormat =
TimeUtils.Formatter.HH_MM_SS;
+
+ /**
+ * Deserializes an array of raw cell bytes into a {@link SeaTunnelRow}.
+ *
+ * @param rawCells one entry per field in {@code rowType}; may be {@code
null} for absent cells
+ * @param rowType the target row schema
+ * @return the deserialized row
+ */
+ public SeaTunnelRow deserialize(ByteString[] rawCells, SeaTunnelRowType
rowType) {
+ SeaTunnelRow row = new SeaTunnelRow(rowType.getTotalFields());
+ for (int i = 0; i < rowType.getTotalFields(); i++) {
+ SeaTunnelDataType<?> fieldType = rowType.getFieldType(i);
+ row.setField(i, deserializeCell(fieldType, rawCells[i]));
+ }
+ return row;
+ }
+
+ private Object deserializeCell(SeaTunnelDataType<?> fieldType, ByteString
cell) {
+ if (cell == null || cell.isEmpty()) {
+ return null;
+ }
+ byte[] bytes = cell.toByteArray();
+ switch (fieldType.getSqlType()) {
+ case TINYINT:
+ return bytes[0];
+ case SMALLINT:
+ return (short) ((bytes[0] & 0xFF) << 8 | (bytes[1] & 0xFF));
+ case INT:
+ return ByteBuffer.wrap(bytes).getInt();
+ case BIGINT:
+ return ByteBuffer.wrap(bytes).getLong();
+ case FLOAT:
+ return ByteBuffer.wrap(bytes).getFloat();
+ case DOUBLE:
+ return ByteBuffer.wrap(bytes).getDouble();
+ case BOOLEAN:
+ return bytes[0] != 0;
+ case BYTES:
+ return bytes;
+ case DECIMAL:
+ String decStr = new String(bytes, StandardCharsets.UTF_8);
+ try {
+ return new BigDecimal(decStr);
+ } catch (NumberFormatException e) {
+ return new BigDecimal(ByteBuffer.wrap(bytes).getFloat());
+ }
+ case DATE:
+ return LocalDate.parse(
+ new String(bytes, StandardCharsets.UTF_8),
+ DateTimeFormatter.ofPattern(dateFormat.getValue()));
+ case TIME:
+ return LocalTime.parse(
+ new String(bytes, StandardCharsets.UTF_8),
+ DateTimeFormatter.ofPattern(timeFormat.getValue()));
+ case TIMESTAMP:
+ return LocalDateTime.parse(
+ new String(bytes, StandardCharsets.UTF_8),
+
DateTimeFormatter.ofPattern(datetimeFormat.getValue()));
+ case STRING:
+ return new String(bytes, StandardCharsets.UTF_8);
+ default:
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.READ_FAILED,
+ "Unsupported data type: " + fieldType.getSqlType());
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSink.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSink.java
new file mode 100644
index 0000000000..468d0f8c52
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSink.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.SchemaSaveMode;
+import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.sink.SupportMultiTableSink;
+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.bigtable.config.BigtableParameters;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableSinkOptions;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.constant.BigtableIdentifier;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorException;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.state.BigtableAggregatedCommitInfo;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.state.BigtableCommitInfo;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.state.BigtableSinkState;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+@Slf4j
+public class BigtableSink
+ implements SeaTunnelSink<
+ SeaTunnelRow,
+ BigtableSinkState,
+ BigtableCommitInfo,
+ BigtableAggregatedCommitInfo>,
+ SupportMultiTableSink {
+
+ private final ReadonlyConfig config;
+ private final CatalogTable catalogTable;
+ private final BigtableParameters parameters;
+ private final SeaTunnelRowType rowType;
+ private final List<Integer> rowkeyColumnIndexes = new ArrayList<>();
+ private int versionColumnIndex = -1;
+ private final SchemaSaveMode schemaSaveMode;
+ private final DataSaveMode dataSaveMode;
+
+ public BigtableSink(ReadonlyConfig config, CatalogTable catalogTable) {
+ this.config = config;
+ this.catalogTable = catalogTable;
+ this.parameters = BigtableParameters.buildWithConfig(config);
+ this.rowType = catalogTable.getSeaTunnelRowType();
+ if (parameters.getVersionColumn() != null) {
+ this.versionColumnIndex =
rowType.indexOf(parameters.getVersionColumn());
+ }
+ this.schemaSaveMode = config.get(BigtableSinkOptions.SCHEMA_SAVE_MODE);
+ this.dataSaveMode = config.get(BigtableSinkOptions.DATA_SAVE_MODE);
+ }
+
+ @Override
+ public String getPluginName() {
+ return BigtableIdentifier.IDENTIFIER_NAME;
+ }
+
+ @Override
+ public BigtableSinkWriter createWriter(SinkWriter.Context context) throws
IOException {
+ for (String rowkeyColumn : parameters.getRowkeyColumns()) {
+ rowkeyColumnIndexes.add(rowType.indexOf(rowkeyColumn));
+ }
+ if (parameters.getVersionColumn() != null) {
+ this.versionColumnIndex =
rowType.indexOf(parameters.getVersionColumn());
+ }
+ handleSaveMode();
+ return new BigtableSinkWriter(rowType, parameters,
rowkeyColumnIndexes, versionColumnIndex);
+ }
+
+ /**
+ * Validates and applies the configured {@link SchemaSaveMode} and {@link
DataSaveMode}.
+ *
+ * <p>Only {@link SchemaSaveMode#RECREATE_SCHEMA} and {@link
DataSaveMode#APPEND_DATA} are
+ * currently supported. Unsupported modes throw immediately so users are
never misled by
+ * accepted-but-no-op settings. Full Admin API support (table creation /
truncation) can be
+ * added in a follow-up once a BigtableCatalog is available.
+ */
+ private void handleSaveMode() {
+ if (schemaSaveMode == SchemaSaveMode.CREATE_SCHEMA_WHEN_NOT_EXIST) {
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.TABLE_CREATE_FAILED,
+ "schema_save_mode=CREATE_SCHEMA_WHEN_NOT_EXIST is not yet
supported by the "
+ + "Bigtable connector. Please create the table and
column families "
+ + "manually and set
schema_save_mode=RECREATE_SCHEMA.");
+ }
+ if (dataSaveMode == DataSaveMode.DROP_DATA) {
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.TABLE_TRUNCATE_FAILED,
+ "data_save_mode=DROP_DATA is not yet supported by the
Bigtable connector. "
+ + "Please truncate the table manually or use
data_save_mode=APPEND_DATA.");
+ }
+ if (dataSaveMode == DataSaveMode.ERROR_WHEN_DATA_EXISTS) {
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.TABLE_QUERY_FAILED,
+ "data_save_mode=ERROR_WHEN_DATA_EXISTS is not yet
supported by the Bigtable "
+ + "connector. Please use
data_save_mode=APPEND_DATA.");
+ }
+ log.info("Bigtable sink save mode: schema={}, data={}",
schemaSaveMode, dataSaveMode);
+ }
+
+ @Override
+ public Optional<CatalogTable> getWriteCatalogTable() {
+ return Optional.ofNullable(catalogTable);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkFactory.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkFactory.java
new file mode 100644
index 0000000000..aed77c0729
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkFactory.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.options.SinkConnectorCommonOptions;
+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.bigtable.config.BigtableBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableSinkOptions;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.constant.BigtableIdentifier;
+
+import com.google.auto.service.AutoService;
+
+@AutoService(Factory.class)
+public class BigtableSinkFactory implements TableSinkFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return BigtableIdentifier.IDENTIFIER_NAME;
+ }
+
+ @Override
+ public OptionRule optionRule() {
+ return OptionRule.builder()
+ .required(
+ BigtableBaseOptions.PROJECT_ID,
+ BigtableBaseOptions.INSTANCE_ID,
+ BigtableBaseOptions.TABLE,
+ BigtableBaseOptions.ROWKEY_COLUMNS,
+ BigtableSinkOptions.COLUMN_FAMILY)
+ .optional(
+ BigtableBaseOptions.CREDENTIALS_PATH,
+ BigtableSinkOptions.ROWKEY_DELIMITER,
+ BigtableSinkOptions.VERSION_COLUMN,
+ BigtableSinkOptions.NULL_MODE,
+ BigtableSinkOptions.BATCH_MUTATION_SIZE,
+ BigtableSinkOptions.SCHEMA_SAVE_MODE,
+ BigtableSinkOptions.DATA_SAVE_MODE,
+ SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA)
+ .build();
+ }
+
+ @Override
+ public TableSink createSink(TableSinkFactoryContext context) {
+ ReadonlyConfig readonlyConfig = context.getOptions();
+ return () -> new BigtableSink(readonlyConfig,
context.getCatalogTable());
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkWriter.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkWriter.java
new file mode 100644
index 0000000000..8ca5312e19
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkWriter.java
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.sink;
+
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.DateTimeUtils;
+import org.apache.seatunnel.common.utils.DateUtils;
+import org.apache.seatunnel.common.utils.TimeUtils;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.client.BigtableClient;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.client.BigtableClient.RowKeyMutation;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableSinkOptions;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorException;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.state.BigtableCommitInfo;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.state.BigtableSinkState;
+
+import com.google.cloud.bigtable.data.v2.models.Mutation;
+import com.google.protobuf.ByteString;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Writes {@link SeaTunnelRow} records to Google Cloud Bigtable.
+ *
+ * <p>Each row is converted to a Bigtable mutation using the configured
column-family mapping and
+ * row-key strategy. Mutations are accumulated in a local buffer and flushed
in bulk when the buffer
+ * reaches {@code batchMutationSize}.
+ */
+@Slf4j
+public class BigtableSinkWriter
+ implements SinkWriter<SeaTunnelRow, BigtableCommitInfo,
BigtableSinkState>,
+ SupportMultiTableSinkWriter<Void> {
+
+ private static final String ALL_COLUMNS_KEY = "all_columns";
+ private static final String DEFAULT_FAMILY = "cf";
+
+ private final BigtableClient bigtableClient;
+ private final SeaTunnelRowType rowType;
+ private final BigtableParameters parameters;
+ private final List<Integer> rowkeyColumnIndexes;
+ private final int versionColumnIndex;
+ private final String defaultFamily;
+
+ /** Buffer of pending mutations, flushed in batches. */
+ private final List<RowKeyMutation> buffer;
+
+ public BigtableSinkWriter(
+ SeaTunnelRowType rowType,
+ BigtableParameters parameters,
+ List<Integer> rowkeyColumnIndexes,
+ int versionColumnIndex) {
+ this(rowType, parameters, rowkeyColumnIndexes, versionColumnIndex,
null);
+ }
+
+ BigtableSinkWriter(
+ SeaTunnelRowType rowType,
+ BigtableParameters parameters,
+ List<Integer> rowkeyColumnIndexes,
+ int versionColumnIndex,
+ BigtableClient bigtableClient) {
+ this.rowType = rowType;
+ this.parameters = parameters;
+ this.rowkeyColumnIndexes = rowkeyColumnIndexes;
+ this.versionColumnIndex = versionColumnIndex;
+ this.buffer = new ArrayList<>(parameters.getBatchMutationSize());
+
+ Map<String, String> familyMap = parameters.getColumnFamily();
+ if (familyMap != null && familyMap.containsKey(ALL_COLUMNS_KEY)) {
+ this.defaultFamily = familyMap.get(ALL_COLUMNS_KEY);
+ } else {
+ this.defaultFamily = DEFAULT_FAMILY;
+ }
+
+ this.bigtableClient =
+ bigtableClient != null ? bigtableClient :
BigtableClient.createInstance(parameters);
+ }
+
+ @Override
+ public void write(SeaTunnelRow element) throws IOException {
+ RowKeyMutation entry = convertRowToMutation(element); // outside lock:
no shared state
+ synchronized (buffer) {
+ buffer.add(entry);
+ if (buffer.size() >= parameters.getBatchMutationSize()) {
+ flush();
+ }
+ }
+ }
+
+ @Override
+ public Optional<BigtableCommitInfo> prepareCommit() throws IOException {
+ synchronized (buffer) {
+ flush();
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public void abortPrepare() {}
+
+ @Override
+ public void close() throws IOException {
+ try {
+ synchronized (buffer) {
+ flush();
+ }
+ } finally {
+ if (bigtableClient != null) {
+ bigtableClient.close();
+ }
+ }
+ }
+
+ private void flush() {
+ if (buffer.isEmpty()) {
+ return;
+ }
+ List<RowKeyMutation> toFlush = new ArrayList<>(buffer);
+ buffer.clear(); // clear first: prevents re-sending if bulkMutate
throws
+ bigtableClient.bulkMutate(toFlush);
+ }
+
+ private RowKeyMutation convertRowToMutation(SeaTunnelRow row) {
+ ByteString rowKey = buildRowKey(row);
+ if (rowKey.isEmpty()) {
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.WRITE_FAILED,
+ "Row key cannot be empty. Check rowkey_column
configuration.");
+ }
+
+ long timestamp = System.currentTimeMillis() * 1000L; // Bigtable uses
microseconds
+ if (versionColumnIndex != -1) {
+ Object versionField = row.getField(versionColumnIndex);
+ if (versionField instanceof Long) {
+ timestamp = (Long) versionField;
+ }
+ }
+
+ Mutation mutation = Mutation.create();
+
+ List<Integer> writeColumnIndexes =
+ IntStream.range(0, row.getArity())
+ .boxed()
+ .filter(idx -> !rowkeyColumnIndexes.contains(idx))
+ .filter(idx -> idx != versionColumnIndex)
+ .collect(Collectors.toList());
+
+ for (Integer idx : writeColumnIndexes) {
+ String fieldName = rowType.getFieldName(idx);
+ String family = resolveFamily(fieldName);
+ Object fieldValue = row.getField(idx);
+ ByteString qualifier = ByteString.copyFromUtf8(fieldName);
+
+ if (fieldValue == null) {
+ if (parameters.getNullMode() ==
BigtableSinkOptions.NullMode.EMPTY) {
+ mutation.setCell(family, qualifier, timestamp,
ByteString.EMPTY);
+ }
+ // SKIP: do nothing
+ } else {
+ ByteString valueBytes = convertToByteString(row, idx);
+ mutation.setCell(family, qualifier, timestamp, valueBytes);
+ }
+ }
+
+ return new RowKeyMutation(rowKey, mutation);
+ }
+
+ private ByteString buildRowKey(SeaTunnelRow row) {
+ if (rowkeyColumnIndexes.size() == 1) {
+ return fieldToByteString(row, rowkeyColumnIndexes.get(0));
+ }
+ String delimiter = parameters.getRowkeyDelimiter();
+ List<String> parts = new ArrayList<>();
+ for (Integer idx : rowkeyColumnIndexes) {
+ Object field = row.getField(idx);
+ parts.add(field == null ? "" : field.toString());
+ }
+ return ByteString.copyFromUtf8(String.join(delimiter, parts));
+ }
+
+ private ByteString fieldToByteString(SeaTunnelRow row, int index) {
+ Object field = row.getField(index);
+ if (field == null) {
+ return ByteString.EMPTY;
+ }
+ SeaTunnelDataType<?> fieldType = rowType.getFieldType(index);
+ if (fieldType.getSqlType() ==
org.apache.seatunnel.api.table.type.SqlType.BYTES) {
+ return ByteString.copyFrom((byte[]) field);
+ }
+ return ByteString.copyFromUtf8(field.toString());
+ }
+
+ private String resolveFamily(String fieldName) {
+ Map<String, String> familyMap = parameters.getColumnFamily();
+ if (familyMap == null) {
+ return defaultFamily;
+ }
+ return familyMap.getOrDefault(fieldName, defaultFamily);
+ }
+
+ private ByteString convertToByteString(SeaTunnelRow row, int index) {
+ Object field = row.getField(index);
+ SeaTunnelDataType<?> fieldType = rowType.getFieldType(index);
+ switch (fieldType.getSqlType()) {
+ case TINYINT:
+ return ByteString.copyFrom(new byte[] {(Byte) field});
+ case SMALLINT:
+ short sv = (Short) field;
+ return ByteString.copyFrom(new byte[] {(byte) (sv >> 8),
(byte) sv});
+ case INT:
+ int iv = (Integer) field;
+ return
ByteString.copyFrom(ByteBuffer.allocate(4).putInt(iv).array());
+ case BIGINT:
+ long lv = (Long) field;
+ return
ByteString.copyFrom(ByteBuffer.allocate(8).putLong(lv).array());
+ case FLOAT:
+ float fv = (Float) field;
+ return
ByteString.copyFrom(ByteBuffer.allocate(4).putFloat(fv).array());
+ case DOUBLE:
+ double dv = (Double) field;
+ return
ByteString.copyFrom(ByteBuffer.allocate(8).putDouble(dv).array());
+ case BOOLEAN:
+ return ByteString.copyFrom(new byte[] {(byte) ((Boolean) field
? 1 : 0)});
+ case BYTES:
+ return ByteString.copyFrom((byte[]) field);
+ case DECIMAL:
+ BigDecimal bd =
+ field instanceof BigDecimal
+ ? (BigDecimal) field
+ : new BigDecimal(field.toString());
+ return
ByteString.copyFrom(bd.toPlainString().getBytes(StandardCharsets.UTF_8));
+ case DATE:
+ LocalDate date =
+ field instanceof LocalDate
+ ? (LocalDate) field
+ : DateUtils.parse(field.toString());
+ return ByteString.copyFrom(
+ DateUtils.toString(date,
DateUtils.Formatter.YYYY_MM_DD)
+ .getBytes(StandardCharsets.UTF_8));
+ case TIME:
+ LocalTime time =
+ field instanceof LocalTime
+ ? (LocalTime) field
+ : TimeUtils.parse(field.toString());
+ return ByteString.copyFrom(
+ TimeUtils.toString(time, TimeUtils.Formatter.HH_MM_SS)
+ .getBytes(StandardCharsets.UTF_8));
+ case TIMESTAMP:
+ LocalDateTime ts =
+ field instanceof LocalDateTime
+ ? (LocalDateTime) field
+ : DateTimeUtils.parse(field.toString());
+ return ByteString.copyFrom(
+ DateTimeUtils.toString(ts,
DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS)
+ .getBytes(StandardCharsets.UTF_8));
+ case STRING:
+ return ByteString.copyFromUtf8(field.toString());
+ default:
+ throw new BigtableConnectorException(
+ BigtableConnectorErrorCode.WRITE_FAILED,
+ String.format(
+ "Bigtable connector does not support column
type [%s]",
+ fieldType.getSqlType()));
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSource.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSource.java
new file mode 100644
index 0000000000..9f9135f68e
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSource.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import org.apache.seatunnel.shade.com.google.common.collect.Lists;
+
+import org.apache.seatunnel.api.source.Boundedness;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import org.apache.seatunnel.api.source.SupportParallelism;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.constant.BigtableIdentifier;
+
+import java.util.List;
+
+public class BigtableSource
+ implements SeaTunnelSource<SeaTunnelRow, BigtableSourceSplit,
BigtableSourceState>,
+ SupportParallelism {
+
+ private final CatalogTable catalogTable;
+ private final BigtableParameters parameters;
+
+ BigtableSource(BigtableParameters parameters, CatalogTable catalogTable) {
+ this.parameters = parameters;
+ this.catalogTable = catalogTable;
+ }
+
+ @Override
+ public String getPluginName() {
+ return BigtableIdentifier.IDENTIFIER_NAME;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public List<CatalogTable> getProducedCatalogTables() {
+ return Lists.newArrayList(catalogTable);
+ }
+
+ @Override
+ public SourceReader<SeaTunnelRow, BigtableSourceSplit> createReader(
+ SourceReader.Context readerContext) throws Exception {
+ return new BigtableSourceReader(
+ parameters, readerContext, catalogTable.getSeaTunnelRowType());
+ }
+
+ @Override
+ public SourceSplitEnumerator<BigtableSourceSplit, BigtableSourceState>
createEnumerator(
+ SourceSplitEnumerator.Context<BigtableSourceSplit>
enumeratorContext) throws Exception {
+ return new BigtableSourceSplitEnumerator(enumeratorContext,
parameters);
+ }
+
+ @Override
+ public SourceSplitEnumerator<BigtableSourceSplit, BigtableSourceState>
restoreEnumerator(
+ SourceSplitEnumerator.Context<BigtableSourceSplit>
enumeratorContext,
+ BigtableSourceState checkpointState)
+ throws Exception {
+ return new BigtableSourceSplitEnumerator(enumeratorContext,
parameters, checkpointState);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceFactory.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceFactory.java
new file mode 100644
index 0000000000..bb4f4d2ad4
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceFactory.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import org.apache.seatunnel.api.configuration.util.OptionRule;
+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.bigtable.config.BigtableBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableSourceOptions;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.constant.BigtableIdentifier;
+
+import com.google.auto.service.AutoService;
+
+import java.io.Serializable;
+
+@AutoService(Factory.class)
+public class BigtableSourceFactory implements TableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return BigtableIdentifier.IDENTIFIER_NAME;
+ }
+
+ @Override
+ public OptionRule optionRule() {
+ return OptionRule.builder()
+ .required(
+ BigtableBaseOptions.PROJECT_ID,
+ BigtableBaseOptions.INSTANCE_ID,
+ BigtableBaseOptions.TABLE)
+ .optional(
+ BigtableBaseOptions.CREDENTIALS_PATH,
+ BigtableBaseOptions.ROWKEY_COLUMNS,
+ BigtableSourceOptions.START_ROW_KEY,
+ BigtableSourceOptions.END_ROW_KEY,
+ BigtableSourceOptions.START_TIMESTAMP,
+ BigtableSourceOptions.END_TIMESTAMP,
+ BigtableSourceOptions.MAX_VERSIONS,
+ BigtableSourceOptions.SCAN_ROW_LIMIT)
+ .build();
+ }
+
+ @Override
+ public Class<? extends SeaTunnelSource> getSourceClass() {
+ return BigtableSource.class;
+ }
+
+ @Override
+ public <T, SplitT extends SourceSplit, StateT extends Serializable>
+ TableSource<T, SplitT, StateT>
createSource(TableSourceFactoryContext context) {
+ return () ->
+ (SeaTunnelSource<T, SplitT, StateT>)
+ new BigtableSource(
+
BigtableParameters.buildWithSourceConfig(context.getOptions()),
+
CatalogTableUtil.buildWithConfig(context.getOptions()));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReader.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReader.java
new file mode 100644
index 0000000000..17e41107aa
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReader.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.client.BigtableClient;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.format.BigtableDeserializationFormat;
+
+import com.google.cloud.bigtable.data.v2.models.Filters;
+import com.google.cloud.bigtable.data.v2.models.Query;
+import com.google.cloud.bigtable.data.v2.models.Row;
+import com.google.cloud.bigtable.data.v2.models.RowCell;
+import com.google.protobuf.ByteString;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentLinkedDeque;
+
+/**
+ * Reads rows from a single {@link BigtableSourceSplit} and emits {@link
SeaTunnelRow} records.
+ *
+ * <p>The column schema is derived from the {@link SeaTunnelRowType}. Each
field name must follow
+ * the pattern {@code columnFamily:qualifier}, except the special name {@code
rowkey} which maps to
+ * the Bigtable row key.
+ */
+@Slf4j
+public class BigtableSourceReader implements SourceReader<SeaTunnelRow,
BigtableSourceSplit> {
+
+ private static final String ROW_KEY_FIELD = "rowkey";
+
+ private final Deque<BigtableSourceSplit> pendingSplits = new
ConcurrentLinkedDeque<>();
+ private final Context context;
+ private final SeaTunnelRowType rowType;
+ private final BigtableParameters parameters;
+ private final BigtableDeserializationFormat deserializationFormat;
+ private volatile boolean noMoreSplits = false;
+ private BigtableClient bigtableClient;
+
+ /** The split currently being read. Persisted in checkpoint to survive
failover. */
+ private volatile BigtableSourceSplit currentSplit = null;
+
+ /**
+ * Set of field names that map to the Bigtable row key. Populated from
{@code rowkey_column}
+ * config; falls back to the literal field name {@value ROW_KEY_FIELD}
when not configured.
+ */
+ private final java.util.Set<String> rowKeyFieldNames;
+
+ public BigtableSourceReader(
+ BigtableParameters parameters, Context context, SeaTunnelRowType
rowType) {
+ this(parameters, context, rowType, null);
+ }
+
+ BigtableSourceReader(
+ BigtableParameters parameters,
+ Context context,
+ SeaTunnelRowType rowType,
+ BigtableClient bigtableClient) {
+ this.parameters = parameters;
+ this.context = context;
+ this.rowType = rowType;
+ this.bigtableClient = bigtableClient;
+ this.deserializationFormat = new BigtableDeserializationFormat();
+ if (parameters.getRowkeyColumns() != null &&
!parameters.getRowkeyColumns().isEmpty()) {
+ this.rowKeyFieldNames = new
java.util.HashSet<>(parameters.getRowkeyColumns());
+ } else {
+ this.rowKeyFieldNames =
java.util.Collections.singleton(ROW_KEY_FIELD);
+ }
+ }
+
+ @Override
+ public void open() throws Exception {
+ if (bigtableClient == null) {
+ bigtableClient = BigtableClient.createInstance(parameters);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (bigtableClient != null) {
+ bigtableClient.close();
+ bigtableClient = null;
+ }
+ }
+
+ @Override
+ public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
+ final BigtableSourceSplit split = pendingSplits.poll();
+ if (Objects.nonNull(split)) {
+ // Assign currentSplit before reading so checkpoints taken during
readSplit()
+ // include it and can re-enqueue it on restore.
+ currentSplit = split;
+ readSplit(split, output);
+ currentSplit = null;
+ } else if (noMoreSplits && pendingSplits.isEmpty()) {
+ log.info("Closed the bounded Bigtable source");
+ context.signalNoMoreElement();
+ } else {
+ log.warn("Waiting for Bigtable split, sleeping 1s");
+ Thread.sleep(1000L);
+ }
+ }
+
+ private void readSplit(BigtableSourceSplit split, Collector<SeaTunnelRow>
output) {
+ Query query = buildQuery(split);
+ // Stream rows one at a time to avoid buffering the full result in
memory.
+ // Each collect() acquires the checkpoint lock only for the emit, not
for the network read.
+ bigtableClient
+ .getDataClient()
+ .readRows(query)
+ .forEach(
+ bigtableRow -> {
+ SeaTunnelRow seaTunnelRow =
convertRow(bigtableRow);
+ synchronized (output.getCheckpointLock()) {
+ output.collect(seaTunnelRow);
+ }
+ });
+ }
+
+ private Query buildQuery(BigtableSourceSplit split) {
+ Query query = Query.create(parameters.getTable());
+
+ String startKey = split.getStartRowKey();
+ String endKey = split.getEndRowKey();
+ if (!startKey.isEmpty() && !endKey.isEmpty()) {
+ query.range(startKey, endKey);
+ } else if (!startKey.isEmpty()) {
+ query.range(startKey, null);
+ } else if (!endKey.isEmpty()) {
+ query.range(null, endKey);
+ }
+
+ if (parameters.getScanRowLimit() > 0) {
+ query.limit(parameters.getScanRowLimit());
+ }
+
+ Filters.Filter filter = buildFilter();
+ if (filter != null) {
+ query.filter(filter);
+ }
+
+ return query;
+ }
+
+ private Filters.Filter buildFilter() {
+ List<Filters.Filter> filters = new ArrayList<>();
+
+ if (parameters.getMaxVersions() > 0) {
+
filters.add(Filters.FILTERS.limit().cellsPerColumn(parameters.getMaxVersions()));
+ }
+
+ Long startTs = parameters.getStartTimestamp();
+ Long endTs = parameters.getEndTimestamp();
+ if (startTs != null && endTs != null) {
+ filters.add(Filters.FILTERS.timestamp().range().of(startTs,
endTs));
+ } else if (startTs != null) {
+
filters.add(Filters.FILTERS.timestamp().range().startClosed(startTs));
+ } else if (endTs != null) {
+ filters.add(Filters.FILTERS.timestamp().range().endOpen(endTs));
+ }
+
+ if (filters.isEmpty()) {
+ return null;
+ }
+ if (filters.size() == 1) {
+ return filters.get(0);
+ }
+ Filters.ChainFilter chain = Filters.FILTERS.chain();
+ for (Filters.Filter f : filters) {
+ chain = chain.filter(f);
+ }
+ return chain;
+ }
+
+ /**
+ * Converts a Bigtable {@link Row} into a {@link SeaTunnelRow}.
+ *
+ * <p>Field names drive the mapping:
+ *
+ * <ul>
+ * <li>Fields listed in {@code rowkey_column} config (or the literal
{@value ROW_KEY_FIELD}
+ * when the option is absent) → the row key bytes
+ * <li>{@code familyName:qualifier} → the latest cell value for that
column
+ * </ul>
+ */
+ private SeaTunnelRow convertRow(Row bigtableRow) {
+ // Build a flat lookup: "family:qualifier" -> latest cell value
+ Map<String, ByteString> cellMap = new HashMap<>();
+ for (RowCell cell : bigtableRow.getCells()) {
+ String key = cell.getFamily() + ":" +
cell.getQualifier().toStringUtf8();
+ cellMap.putIfAbsent(key, cell.getValue()); // first cell = latest
(sorted desc by ts)
+ }
+
+ String[] fieldNames = rowType.getFieldNames();
+ ByteString[] rawCells = new ByteString[fieldNames.length];
+ for (int i = 0; i < fieldNames.length; i++) {
+ String fieldName = fieldNames[i];
+ if (rowKeyFieldNames.contains(fieldName)) {
+ rawCells[i] = bigtableRow.getKey();
+ } else {
+ rawCells[i] = cellMap.get(fieldName);
+ }
+ }
+ return deserializationFormat.deserialize(rawCells, rowType);
+ }
+
+ @Override
+ public List<BigtableSourceSplit> snapshotState(long checkpointId) {
+ List<BigtableSourceSplit> state = new ArrayList<>();
+ // Include the split currently being read so it can be re-enqueued on
restore.
+ if (currentSplit != null) {
+ state.add(currentSplit);
+ }
+ state.addAll(pendingSplits);
+ return state;
+ }
+
+ @Override
+ public void addSplits(List<BigtableSourceSplit> splits) {
+ pendingSplits.addAll(splits);
+ }
+
+ @Override
+ public void handleNoMoreSplits() {
+ noMoreSplits = true;
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {}
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplit.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplit.java
new file mode 100644
index 0000000000..99e6782960
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplit.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import org.apache.seatunnel.api.source.SourceSplit;
+
+import java.io.Serializable;
+
+public class BigtableSourceSplit implements SourceSplit, Serializable {
+
+ private static final long serialVersionUID = 1L;
+ public static final String SPLIT_PREFIX = "bigtable_source_split_";
+
+ private final String splitId;
+ /** Inclusive start row key (empty means the table start). */
+ private final String startRowKey;
+ /** Exclusive end row key (empty means the table end). */
+ private final String endRowKey;
+
+ public BigtableSourceSplit(int splitIndex, String startRowKey, String
endRowKey) {
+ this.splitId = SPLIT_PREFIX + splitIndex;
+ this.startRowKey = startRowKey;
+ this.endRowKey = endRowKey;
+ }
+
+ @Override
+ public String splitId() {
+ return splitId;
+ }
+
+ public String getStartRowKey() {
+ return startRowKey;
+ }
+
+ public String getEndRowKey() {
+ return endRowKey;
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "{\"split_id\":\"%s\", \"start\":\"%s\", \"end\":\"%s\"}",
+ splitId, startRowKey, endRowKey);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumerator.java
new file mode 100644
index 0000000000..3d41a2ad39
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumerator.java
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Enumerates {@link BigtableSourceSplit} instances for parallel reading.
+ *
+ * <p>Currently produces a single split covering the full table (or the
user-defined row-key range).
+ * The split is assigned to whichever reader registers first. Future work can
partition by Bigtable
+ * tablet boundaries using the Admin API.
+ */
+@Slf4j
+public class BigtableSourceSplitEnumerator
+ implements SourceSplitEnumerator<BigtableSourceSplit,
BigtableSourceState> {
+
+ private final Context<BigtableSourceSplit> context;
+ private final BigtableParameters parameters;
+ private final Set<BigtableSourceSplit> assignedSplits;
+ private Set<BigtableSourceSplit> pendingSplits;
+ private boolean initialized = false;
+
+ public BigtableSourceSplitEnumerator(
+ Context<BigtableSourceSplit> context, BigtableParameters
parameters) {
+ this(context, parameters, new HashSet<>());
+ }
+
+ public BigtableSourceSplitEnumerator(
+ Context<BigtableSourceSplit> context,
+ BigtableParameters parameters,
+ BigtableSourceState sourceState) {
+ this(context, parameters, sourceState.getAssignedSplits());
+ }
+
+ private BigtableSourceSplitEnumerator(
+ Context<BigtableSourceSplit> context,
+ BigtableParameters parameters,
+ Set<BigtableSourceSplit> assignedSplits) {
+ this.context = context;
+ this.parameters = parameters;
+ this.assignedSplits = new HashSet<>(assignedSplits);
+ }
+
+ @Override
+ public void open() {
+ this.pendingSplits = new HashSet<>();
+ this.initialized = false;
+ }
+
+ @Override
+ public void run() throws Exception {
+ // Splits are assigned lazily when readers register.
+ }
+
+ @Override
+ public void close() throws IOException {
+ // Nothing to close – no persistent connection held here.
+ }
+
+ @Override
+ public void addSplitsBack(List<BigtableSourceSplit> splits, int subtaskId)
{
+ if (!splits.isEmpty()) {
+ pendingSplits.addAll(splits);
+ if (context.registeredReaders().contains(subtaskId)) {
+ assignSplit(subtaskId);
+ }
+ }
+ }
+
+ @Override
+ public int currentUnassignedSplitSize() {
+ return pendingSplits.size();
+ }
+
+ @Override
+ public void registerReader(int subtaskId) {
+ initializePendingSplits();
+ assignSplit(subtaskId);
+ }
+
+ @Override
+ public BigtableSourceState snapshotState(long checkpointId) throws
Exception {
+ return new BigtableSourceState(assignedSplits);
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {}
+
+ @Override
+ public void handleSplitRequest(int subtaskId) {}
+
+ private void initializePendingSplits() {
+ if (initialized) {
+ return;
+ }
+ Set<BigtableSourceSplit> tableSplits = buildSplits();
+ Set<String> existingIds =
+ pendingSplits.stream()
+ .map(BigtableSourceSplit::splitId)
+ .collect(Collectors.toSet());
+ existingIds.addAll(
+ assignedSplits.stream()
+ .map(BigtableSourceSplit::splitId)
+ .collect(Collectors.toSet()));
+ tableSplits.stream()
+ .filter(s -> !existingIds.contains(s.splitId()))
+ .forEach(pendingSplits::add);
+ initialized = true;
+ }
+
+ /**
+ * Builds the set of splits.
+ *
+ * <p>For now a single split spanning the requested row-key range is
produced. This is
+ * sufficient for bounded batch reads. Parallel multi-split support can be
added later by
+ * querying Bigtable tablet boundary information.
+ */
+ private Set<BigtableSourceSplit> buildSplits() {
+ String startKey = parameters.getStartRowkey() != null ?
parameters.getStartRowkey() : "";
+ String endKey = parameters.getEndRowkey() != null ?
parameters.getEndRowkey() : "";
+ return Collections.singleton(new BigtableSourceSplit(0, startKey,
endKey));
+ }
+
+ private void assignSplit(int taskId) {
+ List<BigtableSourceSplit> toAssign = new ArrayList<>();
+ if (context.currentParallelism() == 1) {
+ toAssign.addAll(pendingSplits);
+ } else {
+ for (BigtableSourceSplit split : pendingSplits) {
+ int owner =
+ (split.splitId().hashCode() & Integer.MAX_VALUE)
+ % context.currentParallelism();
+ if (owner == taskId) {
+ toAssign.add(split);
+ }
+ }
+ }
+ context.assignSplit(taskId, toAssign);
+ assignedSplits.addAll(toAssign);
+ toAssign.forEach(pendingSplits::remove);
+ log.info(
+ "SubTask {} assigned [{}]",
+ taskId,
+ toAssign.stream()
+ .map(BigtableSourceSplit::splitId)
+ .collect(Collectors.joining(",")));
+ context.signalNoMoreSplits(taskId);
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceState.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceState.java
new file mode 100644
index 0000000000..e44fd024a9
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceState.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import java.io.Serializable;
+import java.util.Set;
+
+public class BigtableSourceState implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Set<BigtableSourceSplit> assignedSplits;
+
+ public BigtableSourceState(Set<BigtableSourceSplit> assignedSplits) {
+ this.assignedSplits = assignedSplits;
+ }
+
+ public Set<BigtableSourceSplit> getAssignedSplits() {
+ return assignedSplits;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableAggregatedCommitInfo.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableAggregatedCommitInfo.java
new file mode 100644
index 0000000000..ca9858ae71
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableAggregatedCommitInfo.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.state;
+
+import java.io.Serializable;
+
+public class BigtableAggregatedCommitInfo implements Serializable {
+ private static final long serialVersionUID = 1L;
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableCommitInfo.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableCommitInfo.java
new file mode 100644
index 0000000000..98b10945bd
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableCommitInfo.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.state;
+
+import java.io.Serializable;
+
+public class BigtableCommitInfo implements Serializable {
+ private static final long serialVersionUID = 1L;
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableSinkState.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableSinkState.java
new file mode 100644
index 0000000000..e4d636e7f9
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/main/java/org/apache/seatunnel/connectors/seatunnel/bigtable/state/BigtableSinkState.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.state;
+
+import java.io.Serializable;
+
+public class BigtableSinkState implements Serializable {
+ private static final long serialVersionUID = 1L;
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkSaveModeTest.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkSaveModeTest.java
new file mode 100644
index 0000000000..9d3bd33ab3
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkSaveModeTest.java
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.sink;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.SchemaSaveMode;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.BasicType;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.exception.BigtableConnectorException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Unit tests for {@link BigtableSink#handleSaveMode()}.
+ *
+ * <p>Verifies that unsupported save modes fail fast with a {@link
BigtableConnectorException}
+ * instead of silently no-opping (Issue 3 fix).
+ */
+class BigtableSinkSaveModeTest {
+
+ private CatalogTable catalogTable;
+
+ @BeforeEach
+ void setUp() {
+ catalogTable =
+ CatalogTable.of(
+ TableIdentifier.of("catalog", "database", "table"),
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id",
+ BasicType.STRING_TYPE,
+ (Long) null,
+ false,
+ null,
+ "row key"))
+ .column(
+ PhysicalColumn.of(
+ "cf:name",
+ BasicType.STRING_TYPE,
+ (Long) null,
+ true,
+ null,
+ ""))
+ .build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ "");
+ }
+
+ private ReadonlyConfig buildConfig(String schemaSaveMode, String
dataSaveMode) {
+ Map<String, Object> map = new HashMap<>();
+ map.put("project_id", "p");
+ map.put("instance_id", "i");
+ map.put("table", "t");
+ map.put("rowkey_column", Arrays.asList("id"));
+ map.put("column_family", Collections.singletonMap("all_columns",
"cf"));
+ if (schemaSaveMode != null) {
+ map.put("schema_save_mode", schemaSaveMode);
+ }
+ if (dataSaveMode != null) {
+ map.put("data_save_mode", dataSaveMode);
+ }
+ return ReadonlyConfig.fromMap(map);
+ }
+
+ private BigtableSink newSink() {
+ ReadonlyConfig config = buildConfig("RECREATE_SCHEMA", "APPEND_DATA");
+ return new BigtableSink(config, catalogTable);
+ }
+
+ private static void setField(Object target, String fieldName, Object
value) throws Exception {
+ Field field = BigtableSink.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ /** Invokes the private handleSaveMode guard without creating a writer or
Bigtable client. */
+ private static void invokeHandleSaveMode(BigtableSink sink) throws
Exception {
+ Method method = BigtableSink.class.getDeclaredMethod("handleSaveMode");
+ method.setAccessible(true);
+ try {
+ method.invoke(sink);
+ } catch (InvocationTargetException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw e;
+ }
+ }
+
+ @Test
+ void testSupportedModeDoesNotThrow() {
+ BigtableSink sink = newSink();
+ assertDoesNotThrow(() -> invokeHandleSaveMode(sink));
+ }
+
+ @Test
+ void testDropDataThrows() throws Exception {
+ BigtableSink sink = newSink();
+ setField(sink, "dataSaveMode", DataSaveMode.DROP_DATA);
+ assertThrows(BigtableConnectorException.class, () ->
invokeHandleSaveMode(sink));
+ }
+
+ @Test
+ void testErrorWhenDataExistsThrows() throws Exception {
+ BigtableSink sink = newSink();
+ setField(sink, "dataSaveMode", DataSaveMode.ERROR_WHEN_DATA_EXISTS);
+ assertThrows(BigtableConnectorException.class, () ->
invokeHandleSaveMode(sink));
+ }
+
+ @Test
+ void testCreateSchemaWhenNotExistThrows() throws Exception {
+ BigtableSink sink = newSink();
+ setField(sink, "schemaSaveMode",
SchemaSaveMode.CREATE_SCHEMA_WHEN_NOT_EXIST);
+ assertThrows(BigtableConnectorException.class, () ->
invokeHandleSaveMode(sink));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkWriterTest.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkWriterTest.java
new file mode 100644
index 0000000000..1d8432190d
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/sink/BigtableSinkWriterTest.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.sink;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.client.BigtableClient;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableSinkOptions;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class BigtableSinkWriterTest {
+
+ private BigtableClient mockClient;
+ private SeaTunnelRowType rowType;
+ private BigtableParameters parameters;
+
+ @BeforeEach
+ void setUp() {
+ mockClient = Mockito.mock(BigtableClient.class);
+
+ rowType =
+ new SeaTunnelRowType(
+ new String[] {"id", "name", "age"},
+ new
org.apache.seatunnel.api.table.type.SeaTunnelDataType[] {
+ BasicType.STRING_TYPE, BasicType.STRING_TYPE,
BasicType.INT_TYPE
+ });
+
+ Map<String, String> familyMap = new HashMap<>();
+ familyMap.put("all_columns", "cf");
+
+ parameters =
+ BigtableParameters.builder()
+ .projectId("test-project")
+ .instanceId("test-instance")
+ .table("test-table")
+ .rowkeyColumns(Arrays.asList("id"))
+ .columnFamily(familyMap)
+ .batchMutationSize(10)
+ .nullMode(BigtableSinkOptions.NullMode.SKIP)
+ .build();
+ }
+
+ @Test
+ void testWriteFlushesOnBatchSize() throws IOException {
+ List<Integer> rowkeyIndexes = Arrays.asList(0);
+ BigtableSinkWriter writer =
+ new BigtableSinkWriter(rowType, parameters, rowkeyIndexes, -1,
mockClient);
+
+ // Write batchMutationSize rows to trigger a flush
+ for (int i = 0; i < parameters.getBatchMutationSize(); i++) {
+ SeaTunnelRow row = new SeaTunnelRow(3);
+ row.setField(0, "key-" + i);
+ row.setField(1, "name-" + i);
+ row.setField(2, i);
+ writer.write(row);
+ }
+
+ verify(mockClient, times(1)).bulkMutate(anyList());
+ }
+
+ @Test
+ void testCloseFlushesRemainingRows() throws IOException {
+ List<Integer> rowkeyIndexes = Arrays.asList(0);
+ BigtableSinkWriter writer =
+ new BigtableSinkWriter(rowType, parameters, rowkeyIndexes, -1,
mockClient);
+
+ // Write fewer rows than batch size
+ SeaTunnelRow row = new SeaTunnelRow(3);
+ row.setField(0, "key-1");
+ row.setField(1, "alice");
+ row.setField(2, 30);
+ writer.write(row);
+
+ // No flush yet
+ verify(mockClient, times(0)).bulkMutate(anyList());
+
+ // Close should flush
+ writer.close();
+ verify(mockClient, times(1)).bulkMutate(anyList());
+ verify(mockClient, times(1)).close();
+ }
+
+ @Test
+ void testNullFieldSkipped() throws IOException {
+ List<Integer> rowkeyIndexes = Arrays.asList(0);
+ BigtableSinkWriter writer =
+ new BigtableSinkWriter(rowType, parameters, rowkeyIndexes, -1,
mockClient);
+
+ SeaTunnelRow row = new SeaTunnelRow(3);
+ row.setField(0, "key-null");
+ row.setField(1, null); // should be skipped
+ row.setField(2, 25);
+ writer.write(row);
+ writer.close();
+
+ // Verify a mutation was still produced (null field just omitted)
+ verify(mockClient, times(1)).bulkMutate(anyList());
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReaderTest.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReaderTest.java
new file mode 100644
index 0000000000..d1e8852028
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceReaderTest.java
@@ -0,0 +1,267 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.client.BigtableClient;
+import
org.apache.seatunnel.connectors.seatunnel.bigtable.config.BigtableParameters;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import com.google.api.gax.rpc.ServerStream;
+import com.google.cloud.bigtable.data.v2.BigtableDataClient;
+import com.google.cloud.bigtable.data.v2.models.Query;
+import com.google.cloud.bigtable.data.v2.models.Row;
+import com.google.cloud.bigtable.data.v2.models.RowCell;
+import com.google.protobuf.ByteString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Consumer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link BigtableSourceReader}.
+ *
+ * <p>Covers:
+ *
+ * <ul>
+ * <li>Checkpoint state includes the in-flight split (Issue 1 fix)
+ * <li>Streaming read path — rows emitted via forEach, not buffered (Issue 2
fix)
+ * <li>rowkey_column config drives row-key field mapping (Issue 3 fix)
+ * </ul>
+ */
+class BigtableSourceReaderTest {
+
+ private BigtableClient mockClient;
+ private BigtableDataClient mockDataClient;
+ private SourceReader.Context mockContext;
+ private SeaTunnelRowType rowType;
+ private BigtableParameters parameters;
+
+ @BeforeEach
+ void setUp() {
+ mockClient = mock(BigtableClient.class);
+ mockDataClient = mock(BigtableDataClient.class);
+ when(mockClient.getDataClient()).thenReturn(mockDataClient);
+
+ mockContext = mock(SourceReader.Context.class);
+
+ rowType =
+ new SeaTunnelRowType(
+ new String[] {"rowkey", "cf:name"},
+ new
org.apache.seatunnel.api.table.type.SeaTunnelDataType[] {
+ BasicType.STRING_TYPE, BasicType.STRING_TYPE
+ });
+
+ parameters =
BigtableParameters.builder().projectId("p").instanceId("i").table("t").build();
+ }
+
+ /** Matches production lifecycle: open() initializes the client before
pollNext(). */
+ private BigtableSourceReader createOpenedReader(
+ BigtableParameters params, SeaTunnelRowType type) throws Exception
{
+ BigtableSourceReader reader =
+ new BigtableSourceReader(params, mockContext, type,
mockClient);
+ reader.open();
+ return reader;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void mockReadStream(Runnable duringForEach) {
+ ServerStream<Row> fakeStream = mock(ServerStream.class);
+ Mockito.doAnswer(
+ invocation -> {
+ if (duringForEach != null) {
+ duringForEach.run();
+ }
+ return null;
+ })
+ .when(fakeStream)
+ .forEach(any());
+ when(mockDataClient.readRows(any(Query.class))).thenReturn(fakeStream);
+ }
+
+ //
-------------------------------------------------------------------------
+ // Issue 1: snapshotState must include the currently-being-read split
+ //
-------------------------------------------------------------------------
+
+ /**
+ * When a split is being read (between addSplits and end of readSplit),
snapshotState must
+ * include it so that a failover can re-enqueue it.
+ */
+ @Test
+ void testSnapshotStateIncludesInFlightSplit() throws Exception {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "a", "z");
+ final List<BigtableSourceSplit>[] capturedState = new List[1];
+
+ BigtableSourceReader reader = createOpenedReader(parameters, rowType);
+ reader.addSplits(Collections.singletonList(split));
+
+ mockReadStream(() -> capturedState[0] = reader.snapshotState(1L));
+
+ Collector<SeaTunnelRow> collector = mock(Collector.class);
+ when(collector.getCheckpointLock()).thenReturn(new Object());
+
+ reader.pollNext(collector);
+
+ assertTrue(
+ capturedState[0].stream().anyMatch(s ->
s.splitId().equals(split.splitId())),
+ "snapshotState taken during readSplit() must include the
in-flight split");
+ }
+
+ /**
+ * After readSplit() completes, currentSplit is cleared. A snapshot taken
after that must NOT
+ * re-include the already-finished split.
+ */
+ @Test
+ void testSnapshotStateAfterReadDoesNotDuplicateSplit() throws Exception {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "", "");
+
+ BigtableSourceReader reader = createOpenedReader(parameters, rowType);
+ reader.addSplits(Collections.singletonList(split));
+
+ mockReadStream(null);
+
+ Collector<SeaTunnelRow> collector = mock(Collector.class);
+ when(collector.getCheckpointLock()).thenReturn(new Object());
+
+ reader.pollNext(collector);
+
+ List<BigtableSourceSplit> state = reader.snapshotState(2L);
+ assertTrue(state.isEmpty(), "State after completed read must be
empty");
+ }
+
+ //
-------------------------------------------------------------------------
+ // Issue 2: rows must be emitted via streaming forEach, not buffered
+ //
-------------------------------------------------------------------------
+
+ /**
+ * Verifies that each row is emitted individually via output.collect()
inside the forEach
+ * lambda, rather than being buffered first.
+ */
+ @SuppressWarnings("unchecked")
+ @Test
+ void testRowsEmittedStreamingNotBuffered() throws Exception {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "", "");
+
+ Row fakeRow = mock(Row.class);
+ RowCell cell = mock(RowCell.class);
+ when(cell.getFamily()).thenReturn("cf");
+ when(cell.getQualifier()).thenReturn(ByteString.copyFromUtf8("name"));
+ when(cell.getValue()).thenReturn(ByteString.copyFromUtf8("alice"));
+ when(fakeRow.getCells()).thenReturn(Collections.singletonList(cell));
+ when(fakeRow.getKey()).thenReturn(ByteString.copyFromUtf8("row-1"));
+
+ ServerStream<Row> fakeStream = mock(ServerStream.class);
+ Mockito.doAnswer(
+ invocation -> {
+ Consumer<Row> action = invocation.getArgument(0);
+ action.accept(fakeRow);
+ return null;
+ })
+ .when(fakeStream)
+ .forEach(any());
+ when(mockDataClient.readRows(any(Query.class))).thenReturn(fakeStream);
+
+ BigtableSourceReader reader = createOpenedReader(parameters, rowType);
+ reader.addSplits(Collections.singletonList(split));
+
+ Object lock = new Object();
+ Collector<SeaTunnelRow> collector = mock(Collector.class);
+ when(collector.getCheckpointLock()).thenReturn(lock);
+
+ reader.pollNext(collector);
+
+ ArgumentCaptor<SeaTunnelRow> captor =
ArgumentCaptor.forClass(SeaTunnelRow.class);
+ verify(collector).collect(captor.capture());
+ assertEquals("alice", captor.getValue().getField(1));
+ }
+
+ //
-------------------------------------------------------------------------
+ // Issue 3: rowkey_column config drives field mapping
+ //
-------------------------------------------------------------------------
+
+ /**
+ * When rowkey_column is configured, the named field should receive the
row key value, not the
+ * default literal "rowkey".
+ */
+ @SuppressWarnings("unchecked")
+ @Test
+ void testRowkeyColumnConfigMapsCorrectField() throws Exception {
+ SeaTunnelRowType customRowType =
+ new SeaTunnelRowType(
+ new String[] {"id", "cf:value"},
+ new
org.apache.seatunnel.api.table.type.SeaTunnelDataType[] {
+ BasicType.STRING_TYPE, BasicType.STRING_TYPE
+ });
+
+ BigtableParameters paramsWithRowkeyCol =
+ BigtableParameters.builder()
+ .projectId("p")
+ .instanceId("i")
+ .table("t")
+ .rowkeyColumns(Collections.singletonList("id"))
+ .build();
+
+ Row fakeRow = mock(Row.class);
+ RowCell cell = mock(RowCell.class);
+ when(cell.getFamily()).thenReturn("cf");
+ when(cell.getQualifier()).thenReturn(ByteString.copyFromUtf8("value"));
+ when(cell.getValue()).thenReturn(ByteString.copyFromUtf8("hello"));
+ when(fakeRow.getCells()).thenReturn(Collections.singletonList(cell));
+ when(fakeRow.getKey()).thenReturn(ByteString.copyFromUtf8("my-key"));
+
+ ServerStream<Row> fakeStream = mock(ServerStream.class);
+ Mockito.doAnswer(
+ invocation -> {
+ Consumer<Row> action = invocation.getArgument(0);
+ action.accept(fakeRow);
+ return null;
+ })
+ .when(fakeStream)
+ .forEach(any());
+ when(mockDataClient.readRows(any(Query.class))).thenReturn(fakeStream);
+
+ BigtableSourceReader reader = createOpenedReader(paramsWithRowkeyCol,
customRowType);
+ reader.addSplits(Collections.singletonList(new BigtableSourceSplit(0,
"", "")));
+
+ Object lock = new Object();
+ Collector<SeaTunnelRow> collector = mock(Collector.class);
+ when(collector.getCheckpointLock()).thenReturn(lock);
+
+ reader.pollNext(collector);
+
+ ArgumentCaptor<SeaTunnelRow> captor =
ArgumentCaptor.forClass(SeaTunnelRow.class);
+ verify(collector).collect(captor.capture());
+ assertEquals("my-key", captor.getValue().getField(0));
+ assertEquals("hello", captor.getValue().getField(1));
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumeratorTest.java
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumeratorTest.java
new file mode 100644
index 0000000000..b9d764b34a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-google-bigtable/src/test/java/org/apache/seatunnel/connectors/seatunnel/bigtable/source/BigtableSourceSplitEnumeratorTest.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.bigtable.source;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class BigtableSourceSplitEnumeratorTest {
+
+ @Test
+ void testSingleSplitFullRange() {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "", "");
+ assertEquals("bigtable_source_split_0", split.splitId());
+ assertEquals("", split.getStartRowKey());
+ assertEquals("", split.getEndRowKey());
+ }
+
+ @Test
+ void testSplitWithRowKeyRange() {
+ BigtableSourceSplit split = new BigtableSourceSplit(0, "aaa", "zzz");
+ assertEquals("aaa", split.getStartRowKey());
+ assertEquals("zzz", split.getEndRowKey());
+ }
+}
diff --git a/seatunnel-connectors-v2/pom.xml b/seatunnel-connectors-v2/pom.xml
index 6e0266fa6d..bb47dfb588 100644
--- a/seatunnel-connectors-v2/pom.xml
+++ b/seatunnel-connectors-v2/pom.xml
@@ -77,6 +77,7 @@
<module>connector-tdengine</module>
<module>connector-selectdb-cloud</module>
<module>connector-hbase</module>
+ <module>connector-google-bigtable</module>
<module>connector-rocketmq</module>
<module>connector-amazonsqs</module>
<module>connector-paimon</module>
diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml
index ec8fd7a951..92d25d8aa8 100644
--- a/seatunnel-dist/pom.xml
+++ b/seatunnel-dist/pom.xml
@@ -418,6 +418,12 @@
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-google-bigtable</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.seatunnel</groupId>
<artifactId>connector-datahub</artifactId>
