diff --git a/docs/en/connectors/sink/Assert.md b/docs/en/connectors/sink/Assert.md index 40c143fd8972..3807e6880dda 100644 --- a/docs/en/connectors/sink/Assert.md +++ b/docs/en/connectors/sink/Assert.md @@ -17,8 +17,11 @@ Assert is a sink connector used to validate pipeline output. It checks row count ## Key Features - [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) -- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) +- [x] [batch](../../introduction/concepts/connector-v2-features.md) +- [x] [stream](../../introduction/concepts/connector-v2-features.md) - [x] [support multiple table write](../../introduction/concepts/connector-v2-features.md) +- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md) ## Options @@ -135,6 +138,16 @@ Sink plugin common parameters, please refer to [Sink Common Options](../common-o - `tables_configs` is used for multi-table jobs. The `table_path` value must match the table path carried by the upstream source. - `equals_to` compares the actual field value with the configured expected value. For complex values such as array, map, and row, use the same HOCON value shape as the source data. +:::tip + +The Assert sink is a terminal sink — it has no external system to write to. Use it to validate intermediate results without needing a downstream database. The connector does not interpret `UPDATE` or `DELETE` row kinds as CDC operations; every received row is asserted against the configured rules. If the configured row range, field value, or catalog metadata check fails, the job fails with the matching error message. + +::: + +## Streaming Validation + +Assert works in both `BATCH` and `STREAMING` job modes. Field rules (`NOT_NULL`, `MIN_LENGTH`, `MAX_LENGTH`, etc.) are checked on every row as it arrives at the sink writer. Row count rules (`MIN_ROW` / `MAX_ROW`) are evaluated **exactly once** when the sink writer closes (at job shutdown, savepoint, or failure), against the cumulative row count observed by that writer instance since it was created — not per checkpoint window, and not reset between checkpoints. If you need per-checkpoint-window row-count validation, that requires a source code change (out of scope for a docs update). + ## Example ### Simple @@ -631,6 +644,67 @@ sink { ``` +### Stream Validation With Checkpoint Window + +The example below shows a streaming job that ends with the row count satisfying the cumulative `MIN_ROW` / `MAX_ROW` window (`50 ≤ total rows ≤ 5000`). The check runs once at writer close against the cumulative count, not per checkpoint window. + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 60000 +} + +source { + FakeSource { + row.num = 1000 + schema { + fields { + name = string + age = int + } + } + plugin_output = "stream_data" + } +} + +sink { + Assert { + plugin_input = "stream_data" + rules = + { + row_rules = [ + { + rule_type = MIN_ROW + rule_value = 50 + }, + { + rule_type = MAX_ROW + rule_value = 5000 + } + ], + field_rules = [{ + field_name = age + field_type = int + field_value = [ + { + rule_type = NOT_NULL + }, + { + rule_type = MIN + rule_value = 0 + }, + { + rule_type = MAX + rule_value = 150 + } + ] + }] + } + } +} +``` + ## Changelog diff --git a/docs/en/connectors/sink/GoogleBigtable.md b/docs/en/connectors/sink/GoogleBigtable.md index a57905e2bcf8..8ff42164a50f 100644 --- a/docs/en/connectors/sink/GoogleBigtable.md +++ b/docs/en/connectors/sink/GoogleBigtable.md @@ -4,15 +4,21 @@ import ChangeLog from '../changelog/connector-google-bigtable.md'; > Google Bigtable sink connector +## Support Those Engines + +> SeaTunnel Zeta
+ ## Description Writes data to Google Cloud Bigtable using the native Bigtable Data v2 Java client. -## Key features +## Key Features - [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) - [x] [batch](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) - [x] [support multiple table write](../../introduction/concepts/connector-v2-features.md) +- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md) ## Options @@ -30,7 +36,7 @@ Writes data to Google Cloud Bigtable using the native Bigtable Data v2 Java clie | batch_mutation_size | int | no | 100 | | schema_save_mode | enum | no | RECREATE_SCHEMA | | data_save_mode | enum | no | APPEND_DATA | -| multi_table_sink_replica | int | no | - | +| multi_table_sink_replica | int | no | 1 | | common-options | | no | - | ### project_id [string] @@ -43,13 +49,13 @@ Bigtable instance ID. Example: `"my-bigtable-instance"` ### table [string] -The Bigtable table name to write to. Example: `"my-table"` +The Bigtable table name to write to. Example: `"my-table"`. The connector does not create the Bigtable table; create it (with all required column families) before running the job. ### 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`. +When multiple columns are specified they are joined with `rowkey_delimiter`. With a single row-key column, a null or empty value fails the job with `WRITE_FAILED`. With multiple row-key columns, a null value in any non-last column silently becomes an empty segment in the composed row key (joined by `rowkey_delimiter`); only when the entire composed key is empty does the job fail. ### column_family [config] @@ -70,6 +76,8 @@ column_family { } ``` +Field names that do not appear in the map fall back to the `all_columns` family, or to the default family `cf` if `all_columns` is not configured. + ### credentials_path [string] Path to the Google Cloud service account JSON key file. @@ -109,13 +117,13 @@ Data save mode. Only `APPEND_DATA` is supported now. ### multi_table_sink_replica [int] -The number of sink replicas used for multi-table writing. For details, see [Sink Common Options](../common-options/sink-common-options.md). +The number of sink replicas used for multi-table writing. For details, see [Sink Common Options](../common-options/sink-common-options.md). `multi_table_sink_replica` increases the number of parallel writer replicas within a single sink instance; the target Bigtable table is fixed by the `table` option and is not derived per upstream table. ### common options Sink plugin common parameters, please refer to [Sink Common Options](../common-options/sink-common-options.md) for details. -## Data Types +## Data Type Mapping All SeaTunnel types are supported: @@ -137,15 +145,20 @@ All SeaTunnel types are supported: :::tip -Bigtable does not have relational columns. The sink writes every non-row-key field as a Bigtable cell. The target column family is selected by `column_family`; the Bigtable qualifier is the SeaTunnel field name. +Bigtable does not have relational columns. The sink writes every non-row-key field as a Bigtable cell. The target column family is selected by `column_family`; the Bigtable qualifier is the SeaTunnel field name. The sink treats every upstream row as an unconditional cell mutation, so `UPDATE` / `DELETE` row kinds are not interpreted as CDC operations and overwrite the previous cell under the same `(row key, column family, qualifier)` triple. ::: -## Example +## Task Example ### Basic — Application Default Credentials ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + sink { GoogleBigtable { project_id = "my-gcp-project" @@ -162,6 +175,11 @@ sink { ### Service Account Key File ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + sink { GoogleBigtable { project_id = "my-gcp-project" @@ -181,6 +199,11 @@ sink { ### Multiple Column Families ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + sink { GoogleBigtable { project_id = "my-gcp-project" @@ -200,6 +223,11 @@ sink { ### Use a version column and empty null values ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + sink { GoogleBigtable { project_id = "my-gcp-project" @@ -217,6 +245,52 @@ sink { } ``` +### Streaming write with checkpoint flush + +In streaming mode, the writer flushes the in-memory mutation buffer at every checkpoint. The current `batch_mutation_size` still controls the in-task buffer; checkpoint frequency only affects how often already buffered mutations are sent to Bigtable. + +```hocon +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 30000 +} + +source { + FakeSource { + row.num = 1000 + schema { + fields { + tenant_id = string + event_id = string + event_ts = bigint + event_type = string + payload = string + } + } + plugin_output = "events_stream" + } +} + +sink { + GoogleBigtable { + plugin_input = "events_stream" + 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 = "#" + version_column = "event_ts" + column_family { + all_columns = "data" + event_type = "meta" + } + batch_mutation_size = 200 + } +} +``` + ## Changelog diff --git a/docs/en/connectors/sink/GoogleFirestore.md b/docs/en/connectors/sink/GoogleFirestore.md index 2b653a75a7b4..38f7232107fa 100644 --- a/docs/en/connectors/sink/GoogleFirestore.md +++ b/docs/en/connectors/sink/GoogleFirestore.md @@ -25,8 +25,9 @@ indexes in Google Cloud before running queries that need them. - [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) - [ ] [cdc](../../introduction/concepts/connector-v2-features.md) - [x] [batch](../../introduction/concepts/connector-v2-features.md) -- [ ] [stream](../../introduction/concepts/connector-v2-features.md) +- [x] [stream](../../introduction/concepts/connector-v2-features.md) - [ ] [support multiple table write](../../introduction/concepts/connector-v2-features.md) +- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md) ## Supported DataSource Info @@ -99,14 +100,16 @@ Sink plugin common parameters, please refer to [Sink Common Options](../common-o ## Notes - The connector currently provides a sink only. There is no GoogleFirestore source connector. -- Each sink block writes to one configured collection. It does not switch collections automatically for multi-table input. +- Each sink block writes to one configured collection. It does not switch collections automatically for multi-table input; use one sink block per Firestore collection. - Firestore document IDs are generated automatically. Use another connector or transform before this sink if you need deterministic document IDs. -- The sink does not interpret `UPDATE` or `DELETE` row kinds as CDC operations. +- The sink does not interpret `UPDATE` or `DELETE` row kinds as CDC operations — every row triggers a Firestore `add` call that produces a new document. - Do not put raw service account JSON directly in `credentials`; encode it with Base64 first. -- Field names in the upstream SeaTunnel schema become Firestore document field - names. +- Field names in the upstream SeaTunnel schema become Firestore document field names. +- The connector works in both `BATCH` and `STREAMING` job modes. In the current implementation `FirestoreSinkWriter.write()` calls the Firestore client's `add(...)` once per row and does not buffer or batch rows, so there is no in-memory write buffer to flush at checkpoint boundaries; checkpoint completion does not imply that all previously written rows have reached Firestore. -## Example +## Task Example + +### Batch write of typed rows ```hocon env { @@ -153,6 +156,39 @@ sink { } ``` +### Streaming write with checkpoint interval + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 30000 +} + +source { + FakeSource { + row.num = 100 + schema = { + fields { + c_string = string + c_int = int + c_timestamp = timestamp + } + } + plugin_output = "firestore_stream" + } +} + +sink { + GoogleFirestore { + plugin_input = "firestore_stream" + project_id = "my-gcp-project" + collection = "events" + credentials = "base64-service-account-json" + } +} +``` + ## Changelog diff --git a/docs/en/connectors/sink/Lance.md b/docs/en/connectors/sink/Lance.md index f32a6313f369..113240fc4aa2 100644 --- a/docs/en/connectors/sink/Lance.md +++ b/docs/en/connectors/sink/Lance.md @@ -13,7 +13,11 @@ import ChangeLog from '../changelog/connector-lance.md'; ## Key Features - [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) +- [x] [batch](../../introduction/concepts/connector-v2-features.md) +- [x] [stream](../../introduction/concepts/connector-v2-features.md) - [x] [support multiple table write](../../introduction/concepts/connector-v2-features.md) +- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md) ## Description @@ -88,6 +92,10 @@ The target Lance table name. When it is not set, the connector uses the upstream Controls how Lance writes the dataset. The default is `CREATE`. Use a value supported by Lance `WriteParams.WriteMode`: `CREATE`, `APPEND`, or `OVERWRITE`. +### lance.write.enable.stable.row.ids + +Whether to enable stable row IDs when writing Lance data. The connector reads this option into `LanceSinkConfig.enableStableRowIds` and exposes it via `getEnableStableRowIds()`, but **the value is currently parsed but not yet applied to the underlying Lance `WriteParams`** in `LanceSinkWriter.initializeDataset()` — toggling it has no observable effect on the write path today. This is a known gap pending a follow-up connector change. + ### lance.write.storage.options Passes extra storage parameters to Lance as key-value pairs. @@ -101,9 +109,15 @@ lance.write.storage.options = { } ``` +### multi_table_sink_replica + +Replica count used by the multi-table sink routing mechanism. Increase it when +a single multi-table job writes to many Lance tables and the default single +replica becomes a bottleneck. See [Sink Common Options](../common-options/sink-common-options.md). + ## Data Type Mapping -Lance uses the Apache Arrow type system. The sink creates the Lance schema from the incoming SeaTunnel schema. +Lance uses the Apache Arrow type system. The sink creates the Lance schema from the incoming SeaTunnel schema. The current mapping narrows every integer SeaTunnel type (`TINYINT`, `SMALLINT`, `INT`, `BIGINT`) to Arrow `int32`, so `BIGINT` values outside the signed 32-bit range are truncated. | SeaTunnel Data Type | Lance / Arrow Data Type | |---------------------|-------------------------| @@ -111,19 +125,25 @@ Lance uses the Apache Arrow type system. The sink creates the Lance schema from | TINYINT | int32 | | SMALLINT | int32 | | INT | int32 | -| BIGINT | int32 | +| BIGINT | int32 (values outside the signed 32-bit range are truncated) | | FLOAT | float32 | | DOUBLE | float64 | | DECIMAL | decimal128 | | NULL | null | | BYTES | binary | | DATE | date32 | -| TIME | time32 | -| TIMESTAMP | timestamp | +| TIME | time32 (millisecond precision) | +| TIMESTAMP | timestamp (microsecond, Asia/Shanghai timezone) | | STRING | utf8 | | ARRAY | list | | MAP | map | +:::tip + +The sink does not interpret `UPDATE` / `DELETE` row kinds as CDC operations — every upstream row is appended to the Lance dataset using the configured `lance.write.mode`. In streaming mode, the writer flushes the in-memory row buffer to Lance on every checkpoint. + +::: + ## Task Example ### Write FakeSource Data To Lance @@ -167,6 +187,79 @@ sink { } ``` +### Append Mode With Larger File Fragments + +`APPEND` mode keeps the existing dataset and adds new rows. Increase +`lance.write.max-rows-per-file` and `lance.write.max-bytes-per-file` to reduce +the number of Lance fragments when appending large batches. + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + row.num = 1000000 + schema = { + fields { + c_string = string + c_int = int + } + } + plugin_output = "fake" + } +} + +sink { + Lance { + dataset_path = "/tmp/seatunnel_mnt/lanceTest/lance_sink_table" + namespace_type = "dir" + namespace_id = "root" + table = "lance_sink_table" + lance.write.mode = "APPEND" + lance.write.max-rows-per-file = 100000 + lance.write.max-rows-per-group = 5000 + lance.write.max-bytes-per-file = 134217728 + } +} +``` + +### Streaming Append With Checkpoint Flush + +```hocon +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 30000 +} + +source { + FakeSource { + row.num = 1000 + schema = { + fields { + c_string = string + c_int = int + } + } + plugin_output = "fake_stream" + } +} + +sink { + Lance { + plugin_input = "fake_stream" + dataset_path = "/tmp/seatunnel_mnt/lanceTest/lance_sink_table" + namespace_type = "dir" + namespace_id = "root" + table = "lance_sink_table" + lance.write.mode = "APPEND" + } +} +``` + ## Changelog diff --git a/docs/en/connectors/sink/Typesense.md b/docs/en/connectors/sink/Typesense.md index bf3a2fc0d461..dcc538dafb75 100644 --- a/docs/en/connectors/sink/Typesense.md +++ b/docs/en/connectors/sink/Typesense.md @@ -2,6 +2,12 @@ import ChangeLog from '../changelog/connector-typesense.md'; # Typesense +> Typesense sink connector + +## Support Those Engines + +> SeaTunnel Zeta
+ ## Description Writes SeaTunnel rows to a Typesense collection. The connector can create the target collection @@ -13,6 +19,8 @@ or more primary key fields. - [ ] [Exactly Once](../../introduction/concepts/connector-v2-features.md) - [x] [CDC](../../introduction/concepts/connector-v2-features.md) - [x] [Multiple Table Sink](../../introduction/concepts/connector-v2-features.md) +- [x] [batch](../../introduction/concepts/connector-v2-features.md) +- [x] [stream](../../introduction/concepts/connector-v2-features.md) - [ ] [timer flush](../../introduction/concepts/connector-v2-features.md) ## Options @@ -28,21 +36,21 @@ or more primary key fields. | api_key | string | Yes | - | Typesense API key. | | max_retry_count | int | No | 3 | Maximum retry count for one bulk request. | | max_batch_size | int | No | 10 | Maximum number of documents sent in one bulk request. | -| multi_table_sink_replica | int | No | - | Number of sink replicas used by the common multi-table sink routing mechanism. | +| multi_table_sink_replica | int | No | 1 | Number of sink replicas used by the common multi-table sink routing mechanism. | | common-options | | No | - | Common sink options. | ### hosts [array] -The access address for Typesense, formatted as `host:port`, e.g., `["typesense-01:8108"]`. +The access address for Typesense, formatted as `host:port`, e.g., `["typesense-01:8108"]`. When several nodes are configured, the sink keeps a single client per writer and does not balance writes across them. ### collection [string] -The name of the collection to write to, e.g., "seatunnel". +The name of the collection to write to, e.g., `"seatunnel"`. In multi-table jobs, every table routes to the same configured collection; configure one sink block per target collection if different tables need different destinations. ### primary_keys [array] Primary key fields used to generate the document `id`. When more than one field is configured, -the connector joins their values with `key_delimiter`. +the connector joins their values with `key_delimiter`. Without `primary_keys`, Typesense assigns its own document IDs and the connector behaves as an append-only write. ### key_delimiter [string] @@ -50,15 +58,15 @@ Sets the delimiter for composite keys (default is `_`). ### api_key [string] -The `api_key` for secure access to Typesense. +The `api_key` for secure access to Typesense. Treat this value as a secret and prefer passing it via a job secret or environment variable when running on shared infrastructure. ### max_retry_count [int] -The maximum number of retry attempts for one batch request. +The maximum number of retry attempts for one batch request. The retry predicate is `exception -> true`, so the connector retries on every exception thrown by `typesenseClient.insert(...)` (network errors, timeouts, and Typesense error responses alike) up to `max_retry_count` times with a fixed 200 ms backoff; it does not currently distinguish transient from permanent failures. ### max_batch_size [int] -The maximum number of documents sent in one batch. +The maximum number of documents sent in one batch. Typesense caps each request; keep this value below the Typesense server-side `per_page` limit. ### multi_table_sink_replica [int] @@ -86,11 +94,17 @@ Choose how to handle existing data on the target side before starting the synchr - `APPEND_DATA`: Retains both the database structure and the data. - `ERROR_WHEN_DATA_EXISTS`: Throws an error if data exists. +:::tip + +The connector uses Typesense's bulk import endpoint. `UPDATE` and `DELETE` row kinds are not interpreted as CDC operations — every upstream row is upserted into the target collection based on the generated document `id`. Use `data_save_mode = DROP_DATA` together with a stable `primary_keys` configuration to make repeated jobs behave like upserts rather than appends. + +::: + ## Task Example ### Write Documents With Primary Keys -```bash +```hocon env { parallelism = 1 job.mode = "BATCH" @@ -130,7 +144,7 @@ sink { ### Read From Typesense And Write To Another Collection -```bash +```hocon env { parallelism = 1 job.mode = "BATCH" @@ -176,6 +190,47 @@ sink { } ``` +### Streaming Upsert With Checkpoint Flush + +In streaming mode, the writer buffers up to `max_batch_size` rows or until the next checkpoint, then issues one bulk request. Pair `data_save_mode = DROP_DATA` with a stable `primary_keys` to make every checkpoint produce an idempotent upsert. + +```hocon +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 30000 +} + +source { + FakeSource { + row.num = 1000 + schema { + fields { + company_name = string + num = long + id = string + num_employees = int + flag = boolean + } + } + plugin_output = "typesense_stream" + } +} + +sink { + Typesense { + plugin_input = "typesense_stream" + hosts = ["localhost:8108"] + collection = "typesense_stream_collection" + api_key = "xyz" + primary_keys = ["id"] + max_batch_size = 100 + schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST" + data_save_mode = "DROP_DATA" + } +} +``` + ## Changelog diff --git a/docs/en/connectors/source/GoogleBigtable.md b/docs/en/connectors/source/GoogleBigtable.md index 64e4d2d39c8b..f09218a8da75 100644 --- a/docs/en/connectors/source/GoogleBigtable.md +++ b/docs/en/connectors/source/GoogleBigtable.md @@ -4,20 +4,26 @@ import ChangeLog from '../changelog/connector-google-bigtable.md'; > Google Bigtable source connector +## Support Those Engines + +> SeaTunnel Zeta
+ ## Description Reads data from Google Cloud Bigtable using the native Bigtable Data v2 Java client. -## Key features +## 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) +- [ ] [column projection](../../introduction/concepts/connector-v2-features.md) - [ ] [parallelism](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) :::tip -The source is bounded. It creates one split for the configured table or row-key range, so increasing job parallelism does not split one Bigtable scan into multiple tablet-range reads. +The source is bounded. It creates one split for the configured table or row-key range, so increasing job parallelism does not split one Bigtable scan into multiple tablet-range reads. Each scan reads every requested cell for the configured row range and emits one SeaTunnel row per Bigtable row. ::: @@ -52,23 +58,27 @@ 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. +Path to the Google Cloud service account JSON key file. If omitted, Application Default Credentials (ADC) are used. ADC works automatically on GCE/GKE nodes, in `gcloud` shell sessions, or when the `GOOGLE_APPLICATION_CREDENTIALS` environment variable points to a service account JSON file. ### rowkey_column [list] -Optional list of field names that should receive the row key value. If this option is not set, the connector uses a schema field named `rowkey` as the row-key field. The row-key field can be declared as `BYTES` or `STRING`. +Optional list of field names that should receive the row key value. If this option is not set, the connector uses a schema field named `rowkey` as the row-key field. + +Each listed field is decoded independently according to its own declared type in `schema.fields`: `BYTES` receives the raw row-key bytes; `STRING` receives a UTF-8 decoded view. Different row-key fields can therefore use different types in the same scan (for example one field exposing the raw key bytes for downstream binary processing, another exposing a UTF-8 view). ### start_rowkey [string] Inclusive start row key for the scan. If not set, the scan starts from the beginning of the table. +The connector passes the value to the Bigtable client as a UTF-8 string; only lexicographic comparison is supported. Use `BYTES` for binary row keys that do not encode as UTF-8. + ### 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). +Inclusive start timestamp filter (microseconds since epoch). Combined with `end_timestamp` and `max_versions`, this controls which cell versions Bigtable returns for each column qualifier. ### end_timestamp [long] @@ -76,11 +86,11 @@ 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. +Maximum number of cell versions to return per column qualifier. Default `1` returns only the latest version. Larger values expose historical cell versions; the source still emits one row per Bigtable row, so older versions of the same cell are flattened into the latest returned cell. ### scan_row_limit [int] -Maximum number of rows to return. `-1` (default) means no limit. +Maximum number of rows to return. `-1` (default) means no limit. Use this option together with `start_rowkey` / `end_rowkey` to do paginated full-table scans across multiple jobs. ### common options @@ -102,11 +112,16 @@ The source reads the latest returned cell for each `family:qualifier` field. Use ::: -## Example +## Task Example -### Read all rows — Application Default Credentials +### Read all rows with Application Default Credentials ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + source { GoogleBigtable { project_id = "my-gcp-project" @@ -126,6 +141,11 @@ source { ### Scan a row-key range with a service account ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + source { GoogleBigtable { project_id = "my-gcp-project" @@ -149,6 +169,11 @@ source { ### Use a custom row-key field name ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + source { GoogleBigtable { project_id = "my-gcp-project" @@ -166,6 +191,38 @@ source { } ``` +### Bounded streaming scan with cell-version filtering + +Use `STREAMING` job mode when you want the scan to run with checkpointing while still being a single bounded read. Combine `start_timestamp`, `end_timestamp`, and `max_versions` to restrict which cell versions Bigtable returns. + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 60000 +} + +source { + GoogleBigtable { + project_id = "my-gcp-project" + instance_id = "my-bigtable-instance" + table = "events" + start_timestamp = 1704067200000000 + end_timestamp = 1735689600000000 + max_versions = 3 + scan_row_limit = 500000 + schema { + fields { + rowkey = STRING + "cf:type" = STRING + "cf:data" = STRING + "cf:ts" = BIGINT + } + } + } +} +``` + ## Changelog diff --git a/docs/en/connectors/source/Typesense.md b/docs/en/connectors/source/Typesense.md index 715838f5c5a3..e1b24a8d51d2 100644 --- a/docs/en/connectors/source/Typesense.md +++ b/docs/en/connectors/source/Typesense.md @@ -4,16 +4,23 @@ import ChangeLog from '../changelog/connector-typesense.md'; > Typesense Source Connector +## Support Those Engines + +> SeaTunnel Zeta
+ ## Description Reads documents from a Typesense collection. The source supports bounded batch reads and can pass Typesense search parameters through `query`. +The source is bounded. Each job reads every document that matches the configured `query` once and then completes. Use Typesense's own change tracking on the receiver side if you need change-data capture. + ## Key Features - [x] [Batch Processing](../../introduction/concepts/connector-v2-features.md) - [ ] [Stream Processing](../../introduction/concepts/connector-v2-features.md) - [ ] [Exactly-Once](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) - [x] [Schema](../../introduction/concepts/connector-v2-features.md) - [x] [Parallelism](../../introduction/concepts/connector-v2-features.md) - [ ] [User-Defined Splits Support](../../introduction/concepts/connector-v2-features.md) @@ -34,7 +41,7 @@ Typesense search parameters through `query`. ### hosts [array] The access address of Typesense. Use the `host:port` format, for example: -`["typesense-01:8108"]`. Multiple hosts are supported. +`["typesense-01:8108"]`. Multiple hosts are supported. When several nodes are configured, the source issues its search requests to the first reachable node; the list is not used for parallel scan sharding. ### collection [string] @@ -46,7 +53,7 @@ The columns to be read from Typesense. For more information, please refer to the ### api_key [string] -The `api_key` for Typesense security authentication. +The `api_key` for Typesense security authentication. Treat this value as a secret and prefer passing it via a job secret or environment variable when running on shared infrastructure. ### protocol [string] @@ -58,9 +65,11 @@ Typesense Cloud or other TLS-enabled endpoints. Typesense search parameters, for example `q=*&filter_by=num_employees:>9000`. If it is not set, the source reads all documents returned by the default search. +Any valid Typesense search parameter can be appended, including `q`, `query_by`, `filter_by`, `sort_by`, `page`, and `per_page`. The connector forwards them to the Typesense search API unchanged. + ### batch_size [int] -The number of records to query per batch when reading data. +The number of records to query per batch when reading data. Each request uses the Typesense `per_page` parameter, so the value must be between 1 and the Typesense server-side `per_page` limit (typically 250). Lower the value if you see truncated pages in the logs. ### Common Options @@ -70,7 +79,7 @@ For common parameters of Source plugins, please refer to [Source Common Options] ### Read Documents With A Filter -```bash +```hocon env { parallelism = 1 job.mode = "BATCH" @@ -105,6 +114,39 @@ sink { } ``` +### Read A Subset With A Custom Query + +Combine `query_by` and `sort_by` to control which fields Typesense searches and how the result set is ordered. + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + Typesense { + hosts = ["localhost:8108"] + collection = "companies" + api_key = "xyz" + query = "q=acme&query_by=company_name&filter_by=country:=US&sort_by=num_employees:desc" + batch_size = 50 + schema = { + fields { + company_name = string + num_employees = long + country = string + id = string + } + } + } +} + +sink { + Console {} +} +``` + ## Changelog diff --git a/docs/zh/connectors/sink/Assert.md b/docs/zh/connectors/sink/Assert.md index 4565d908d960..ef5c5efea2f0 100644 --- a/docs/zh/connectors/sink/Assert.md +++ b/docs/zh/connectors/sink/Assert.md @@ -17,8 +17,11 @@ Assert 是一个用于校验任务输出结果的数据接收器。它可以按 ## 主要特性 - [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) -- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) +- [x] [批处理](../../introduction/concepts/connector-v2-features.md) +- [x] [流处理](../../introduction/concepts/connector-v2-features.md) - [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md) +- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md) ## 配置 @@ -136,6 +139,16 @@ Sink 插件的通用参数，请参考 [Sink Common Options](../common-options/s - `tables_configs` 用于多表任务，`table_path` 必须和上游数据携带的表路径一致。 - `equals_to` 会比较实际字段值和配置的期望值。数组、Map、Row 这类复杂值需要使用和 source 数据一致的 HOCON 写法。 +:::tip + +Assert 是一个终端 sink —— 没有外部存储系统可以写入。它适合在不需要下游数据库的情况下校验中间结果。连接器不会按 `UPDATE` 或 `DELETE` 行类型执行 CDC 语义，每条收到的记录都会按配置的规则进行断言。行数、字段值或 catalog 元数据校验失败时，任务会以对应的错误信息直接失败。 + +::: + +## 流式校验 + +Assert 同时支持 `BATCH` 与 `STREAMING` 两种作业模式。字段规则（如 `NOT_NULL`、`MIN_LENGTH`、`MAX_LENGTH` 等）会在每一条记录到达 Sink Writer 时进行检查；行数规则（`MIN_ROW` / `MAX_ROW`）只在 Sink Writer 关闭时（作业关闭、savepoint 或失败时）执行一次，对比的是该 Writer 实例自创建以来累计接收的总行数，既不会按 checkpoint 窗口重复校验，也不会在 checkpoint 之间重置。如果需要真正按 checkpoint 窗口的行数校验，这超出了文档更新范围，需要改动源代码。 + ## 示例 ### 简单 @@ -632,6 +645,67 @@ sink { ``` +### 流式校验并按 Checkpoint 窗口断言行数 + +下面的示例演示一个流式作业：作业结束时累计行数满足 `MIN_ROW` / `MAX_ROW` 区间（`50 ≤ 总行数 ≤ 5000`）。该校验只在 Writer 关闭时执行一次，对比累计行数，并不是按 checkpoint 窗口重复执行。 + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 60000 +} + +source { + FakeSource { + row.num = 1000 + schema = { + fields { + name = string + age = int + } + } + plugin_output = "stream_data" + } +} + +sink { + Assert { + plugin_input = "stream_data" + rules = + { + row_rules = [ + { + rule_type = MIN_ROW + rule_value = 50 + }, + { + rule_type = MAX_ROW + rule_value = 5000 + } + ], + field_rules = [{ + field_name = age + field_type = int + field_value = [ + { + rule_type = NOT_NULL + }, + { + rule_type = MIN + rule_value = 0 + }, + { + rule_type = MAX + rule_value = 150 + } + ] + }] + } + } +} +``` + ## 变更日志 diff --git a/docs/zh/connectors/sink/GoogleBigtable.md b/docs/zh/connectors/sink/GoogleBigtable.md index 4d05bc81bedf..ba3a7e020d7c 100644 --- a/docs/zh/connectors/sink/GoogleBigtable.md +++ b/docs/zh/connectors/sink/GoogleBigtable.md @@ -4,15 +4,21 @@ import ChangeLog from '../changelog/connector-google-bigtable.md'; > Google Bigtable Sink 连接器 +## 支持这些引擎 + +> SeaTunnel Zeta
+ ## 描述 使用原生 Bigtable Data v2 Java 客户端将数据写入 Google Cloud Bigtable。 ## 主要特性 -- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) +- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) - [x] [批处理](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) - [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md) +- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md) ## 参数 @@ -30,7 +36,7 @@ import ChangeLog from '../changelog/connector-google-bigtable.md'; | batch_mutation_size| int | 否 | 100 | | schema_save_mode | enum | 否 | RECREATE_SCHEMA | | data_save_mode | enum | 否 | APPEND_DATA | -| multi_table_sink_replica | int | 否 | - | +| multi_table_sink_replica | int | 否 | 1 | | common-options | | 否 | - | ### project_id [string] @@ -43,11 +49,11 @@ Bigtable 实例 ID，例如 `"my-bigtable-instance"`。 ### table [string] -写入的 Bigtable 表名，例如 `"my-table"`。 +写入的 Bigtable 表名，例如 `"my-table"`。连接器不会自动建表，需要先在 Bigtable 中创建好目标表以及会用到的列族。 ### rowkey_column [list] -用于构造行键的列名列表，例如 `["id"]` 或 `["tenant_id", "event_id"]`。多列时用 `rowkey_delimiter` 拼接。 +用于构造行键的列名列表，例如 `["id"]` 或 `["tenant_id", "event_id"]`。多列时用 `rowkey_delimiter` 拼接。当只有一个行键列时，值为 `null` 或空字符串会让作业直接以 `WRITE_FAILED` 失败；当配置了多个行键列时，非末尾列为 `null` 会被静默转成空串并通过 `rowkey_delimiter` 拼接到组合行键中，只有当整条组合行键最终为空时作业才会失败。 ### column_family [config] @@ -68,6 +74,8 @@ column_family { } ``` +未在映射中出现的字段名会回退到 `all_columns` 指定的列族；如果也没有 `all_columns`，则使用默认列族 `cf`。 + ### credentials_path [string] Google Cloud 服务账号 JSON 密钥文件路径。未设置时使用应用默认凭证（ADC）。 @@ -102,13 +110,13 @@ Schema 保存模式。当前只支持 `RECREATE_SCHEMA`。 ### multi_table_sink_replica [int] -多表写入时使用的 Sink 副本数。更多说明请参考 [Sink Common Options](../common-options/sink-common-options.md)。 +多表写入时使用的 Sink 副本数。`multi_table_sink_replica` 用于在单个 Sink 实例中增加并行写入副本数；目标 Bigtable 表由 `table` 选项固定，不会根据上游表名动态切换。更多说明请参考 [Sink Common Options](../common-options/sink-common-options.md)。 ### common options Sink 插件通用参数，详见 [Sink Common Options](../common-options/sink-common-options.md)。 -## 数据类型 +## 数据类型映射 Bigtable 没有关系型数据库那样的列类型。连接器会按如下格式把 SeaTunnel 字段写入 Bigtable Cell： @@ -130,15 +138,20 @@ Bigtable 没有关系型数据库那样的列类型。连接器会按如下格 :::tip -Sink 会把非行键字段写成 Bigtable Cell。目标列族由 `column_family` 决定，Bigtable 的列限定符使用 SeaTunnel 字段名。 +Sink 会把非行键字段写成 Bigtable Cell。目标列族由 `column_family` 决定，Bigtable 的列限定符使用 SeaTunnel 字段名。每条上游记录都会被当成无条件的 Cell 变更，所以 `UPDATE` / `DELETE` 类型的行不会被解释为 CDC 操作，而是直接覆盖相同 `(行键, 列族, 列限定符)` 下的旧 Cell。 ::: -## 示例 +## 任务示例 ### 使用应用默认凭证写入 ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + sink { GoogleBigtable { project_id = "my-gcp-project" @@ -155,6 +168,11 @@ sink { ### 使用服务账号和复合行键写入 ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + sink { GoogleBigtable { project_id = "my-gcp-project" @@ -174,6 +192,11 @@ sink { ### 写入多个列族 ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + sink { GoogleBigtable { project_id = "my-gcp-project" @@ -193,6 +216,11 @@ sink { ### 使用版本列并把空值写成空字节 ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + sink { GoogleBigtable { project_id = "my-gcp-project" @@ -210,6 +238,52 @@ sink { } ``` +### 流式写入并按 Checkpoint 刷新 + +在流式模式下，Writer 会在每次 checkpoint 触发时把本地 mutation 缓冲写入 Bigtable。`batch_mutation_size` 仍然控制任务内部缓冲，checkpoint 频率只会影响已缓冲的 mutation 多久被发送到 Bigtable。 + +```hocon +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 30000 +} + +source { + FakeSource { + row.num = 1000 + schema { + fields { + tenant_id = string + event_id = string + event_ts = bigint + event_type = string + payload = string + } + } + plugin_output = "events_stream" + } +} + +sink { + GoogleBigtable { + plugin_input = "events_stream" + 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 = "#" + version_column = "event_ts" + column_family { + all_columns = "data" + event_type = "meta" + } + batch_mutation_size = 200 + } +} +``` + ## Changelog diff --git a/docs/zh/connectors/sink/GoogleFirestore.md b/docs/zh/connectors/sink/GoogleFirestore.md index 40133e5fb50d..2d948adddd18 100644 --- a/docs/zh/connectors/sink/GoogleFirestore.md +++ b/docs/zh/connectors/sink/GoogleFirestore.md @@ -24,8 +24,9 @@ GoogleFirestore Sink 用于将 SeaTunnel 数据写入 Google Cloud Firestore 集 - [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) - [ ] [CDC](../../introduction/concepts/connector-v2-features.md) - [x] [批处理](../../introduction/concepts/connector-v2-features.md) -- [ ] [流处理](../../introduction/concepts/connector-v2-features.md) +- [x] [流处理](../../introduction/concepts/connector-v2-features.md) - [ ] [支持多表写入](../../introduction/concepts/connector-v2-features.md) +- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md) ## 支持的数据源信息 @@ -97,13 +98,16 @@ Sink 插件通用参数，请参考 [Sink 通用选项](../common-options/sink-c ## 注意事项 - 当前连接器只提供 sink，不提供 GoogleFirestore source。 -- 每个 sink 配置块只写入一个固定的 collection，不会按多表输入自动切换 collection。 +- 每个 sink 配置块只写入一个固定的 collection，不会按多表输入自动切换 collection；多 collection 场景需要为每个目标 collection 单独配置一个 sink 块。 - Firestore 文档 ID 会自动生成。如果需要固定文档 ID，请在写入前使用其他连接器或转换处理。 -- sink 不会按 `UPDATE` 或 `DELETE` 行类型执行 CDC 语义。 +- sink 不会按 `UPDATE` 或 `DELETE` 行类型执行 CDC 语义 —— 每条记录都会触发一次 Firestore `add` 调用生成新文档。 - `credentials` 不能直接填写服务账号 JSON 原文，需要先做 Base64 编码。 - 上游 SeaTunnel schema 中的字段名会作为 Firestore 文档字段名。 +- 连接器同时支持 `BATCH` 与 `STREAMING` 两种作业模式。在当前实现中，`FirestoreSinkWriter.write()` 对每一条记录直接调用一次 Firestore 客户端的 `add(...)`，并不会缓冲或批量写入，因此并没有“checkpoint 前把内存写缓冲刷到 Firestore”这一行为；checkpoint 完成并不意味着之前调用过的所有写入都已真正到达 Firestore。 -## 示例 +## 任务示例 + +### 批量写入带类型的记录 ```hocon env { @@ -150,6 +154,39 @@ sink { } ``` +### 流式写入并启用 Checkpoint 间隔 + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 30000 +} + +source { + FakeSource { + row.num = 100 + schema = { + fields { + c_string = string + c_int = int + c_timestamp = timestamp + } + } + plugin_output = "firestore_stream" + } +} + +sink { + GoogleFirestore { + plugin_input = "firestore_stream" + project_id = "my-gcp-project" + collection = "events" + credentials = "base64-service-account-json" + } +} +``` + ## 变更日志 diff --git a/docs/zh/connectors/sink/Lance.md b/docs/zh/connectors/sink/Lance.md index 15350b09055e..59c478fbb8d0 100644 --- a/docs/zh/connectors/sink/Lance.md +++ b/docs/zh/connectors/sink/Lance.md @@ -13,7 +13,11 @@ import ChangeLog from '../changelog/connector-lance.md'; ## 主要特性 - [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) +- [x] [批处理](../../introduction/concepts/connector-v2-features.md) +- [x] [流处理](../../introduction/concepts/connector-v2-features.md) - [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md) +- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md) ## 描述 @@ -82,6 +86,10 @@ Lance namespace 的根目录。SeaTunnel 运行用户需要有权限在该目录 控制 Lance 的写入方式。默认值是 `CREATE`。该值需要是 Lance `WriteParams.WriteMode` 支持的值：`CREATE`、`APPEND` 或 `OVERWRITE`。 +### lance.write.enable.stable.row.ids + +写入 Lance 时是否启用稳定的 row ID。连接器会把这个选项读入 `LanceSinkConfig.enableStableRowIds`，并通过 `getEnableStableRowIds()` 暴露，但**当前实现中该值仅被解析，还未真正传入底层的 Lance `WriteParams`**（`LanceSinkWriter.initializeDataset()` 构造的 `WriteParams` 不包含这个开关），目前切换它对写入路径没有可见效果。这是一项已知的缺口，需要后续连接器提交来补齐。 + ### lance.write.storage.options 以键值对形式传递额外的 Lance 存储参数。 @@ -95,9 +103,13 @@ lance.write.storage.options = { } ``` +### multi_table_sink_replica + +多表写入时的 sink 并行副本数。当一个多表作业写入大量 Lance 表、单个副本成为瓶颈时调大该值。详见 [Sink 通用选项](../common-options/sink-common-options.md)。 + ## 数据类型映射 -Lance 使用 Apache Arrow 类型系统。sink 会根据上游 SeaTunnel 表结构创建 Lance schema。 +Lance 使用 Apache Arrow 类型系统。sink 会根据上游 SeaTunnel 表结构创建 Lance schema。当前映射会把所有整数类型（`TINYINT`、`SMALLINT`、`INT`、`BIGINT`）一律收窄为 Arrow `int32`，因此超出有符号 32 位范围的 `BIGINT` 值会被截断。 | SeaTunnel 数据类型 | Lance / Arrow 数据类型 | |--------------------|------------------------| @@ -105,19 +117,25 @@ Lance 使用 Apache Arrow 类型系统。sink 会根据上游 SeaTunnel 表结 | TINYINT | int32 | | SMALLINT | int32 | | INT | int32 | -| BIGINT | int32 | +| BIGINT | int32（超出有符号 32 位范围的值会被截断） | | FLOAT | float32 | | DOUBLE | float64 | | DECIMAL | decimal128 | | NULL | null | | BYTES | binary | | DATE | date32 | -| TIME | time32 | -| TIMESTAMP | timestamp | +| TIME | time32（毫秒精度） | +| TIMESTAMP | timestamp（微秒精度，Asia/Shanghai 时区） | | STRING | utf8 | | ARRAY | list | | MAP | map | +:::tip + +Sink 不会按 `UPDATE` / `DELETE` 行类型执行 CDC 语义 —— 每条上游记录都会按 `lance.write.mode` 追加到 Lance 数据集中。在流式模式下，Writer 会在每个 checkpoint 把内存中的行缓冲写入 Lance。 + +::: + ## 任务示例 ### 写入 FakeSource 数据到 Lance @@ -161,6 +179,77 @@ sink { } ``` +### 使用 APPEND 模式并调大文件分片 + +`APPEND` 模式会保留已有数据集并写入新行。把 `lance.write.max-rows-per-file` 和 `lance.write.max-bytes-per-file` 调大，可以减少追加大批量数据时产生的 Lance fragment 数量。 + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + row.num = 1000000 + schema = { + fields { + c_string = string + c_int = int + } + } + plugin_output = "fake" + } +} + +sink { + Lance { + dataset_path = "/tmp/seatunnel_mnt/lanceTest/lance_sink_table" + namespace_type = "dir" + namespace_id = "root" + table = "lance_sink_table" + lance.write.mode = "APPEND" + lance.write.max-rows-per-file = 100000 + lance.write.max-rows-per-group = 5000 + lance.write.max-bytes-per-file = 134217728 + } +} +``` + +### 流式追加并按 Checkpoint 刷新 + +```hocon +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 30000 +} + +source { + FakeSource { + row.num = 1000 + schema = { + fields { + c_string = string + c_int = int + } + } + plugin_output = "fake_stream" + } +} + +sink { + Lance { + plugin_input = "fake_stream" + dataset_path = "/tmp/seatunnel_mnt/lanceTest/lance_sink_table" + namespace_type = "dir" + namespace_id = "root" + table = "lance_sink_table" + lance.write.mode = "APPEND" + } +} +``` + ## 更新日志 diff --git a/docs/zh/connectors/sink/Typesense.md b/docs/zh/connectors/sink/Typesense.md index faca40ae01c8..c9217c8dd3bb 100644 --- a/docs/zh/connectors/sink/Typesense.md +++ b/docs/zh/connectors/sink/Typesense.md @@ -2,6 +2,12 @@ import ChangeLog from '../changelog/connector-typesense.md'; # Typesense +> Typesense sink 连接器 + +## 支持的引擎 + +> SeaTunnel Zeta
+ ## 描述 将 SeaTunnel 数据写入 Typesense collection。该 connector 可以按配置创建目标 collection、 @@ -12,6 +18,8 @@ import ChangeLog from '../changelog/connector-typesense.md'; - [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) - [x] [CDC](../../introduction/concepts/connector-v2-features.md) - [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md) +- [x] [批处理](../../introduction/concepts/connector-v2-features.md) +- [x] [流处理](../../introduction/concepts/connector-v2-features.md) - [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md) ## 选项 @@ -27,20 +35,20 @@ import ChangeLog from '../changelog/connector-typesense.md'; | api_key | string | 是 | - | Typesense API Key。 | | max_retry_count | int | 否 | 3 | 单个批量请求的最大重试次数。 | | max_batch_size | int | 否 | 10 | 单个批量请求最多写入的文档数量。 | -| multi_table_sink_replica | int | 否 | - | 通用多表写入路由机制使用的 Sink 副本数。 | +| multi_table_sink_replica | int | 否 | 1 | 通用多表写入路由机制使用的 Sink 副本数。 | | common-options | | 否 | - | 通用 Sink 选项。 | ### hosts [array] -Typesense 的访问地址，格式为 `host:port`，例如：`["typesense-01:8108"]`。 +Typesense 的访问地址，格式为 `host:port`，例如：`["typesense-01:8108"]`。配置多个节点时，每个 Writer 只持有一个客户端，不会把写入请求在节点之间负载均衡。 ### collection [string] -要写入的 collection 名，例如：`seatunnel`。 +要写入的 collection 名，例如：`seatunnel`。在多表作业中，所有表都会路由到同一个 collection；如果不同表要写入不同目标，请为每个目标 collection 单独配置一个 sink 块。 ### primary_keys [array] -主键字段用于生成文档 `id`。配置多个字段时，connector 会用 `key_delimiter` 拼接这些字段值。 +主键字段用于生成文档 `id`。配置多个字段时，connector 会用 `key_delimiter` 拼接这些字段值。未配置 `primary_keys` 时，Typesense 会自行分配文档 ID，连接器退化为纯追加写入。 ### key_delimiter [string] @@ -48,15 +56,15 @@ Typesense 的访问地址，格式为 `host:port`，例如：`["typesense-01:810 ### api_key [string] -Typesense 安全认证的 `api_key`。 +Typesense 安全认证的 `api_key`。请把它当作敏感凭据处理；在共享环境运行时，建议通过作业密钥或环境变量注入。 ### max_retry_count [int] -单个批量请求的最大重试次数。 +单个批量请求的最大重试次数。重试谓词为 `exception -> true`，也就是说 `typesenseClient.insert(...)` 抛出的任何异常（网络错误、超时以及 Typesense 业务错误响应）都会被同样重试，最多执行 `max_retry_count` 次，每次间隔固定的 200 ms；当前实现并不会区分瞬时错误和永久错误。 ### max_batch_size [int] -每批最多写入的文档数量。 +每批最多写入的文档数量。Typesense 对单次请求有上限，请将该值保持在 Typesense 服务端 `per_page` 上限以下。 ### multi_table_sink_replica [int] @@ -85,11 +93,17 @@ Typesense collection 创建时会使用上游 SeaTunnel 表结构。如果希望 `APPEND_DATA`：保留数据库结构，保留数据
`ERROR_WHEN_DATA_EXISTS`：当有数据时抛出错误
+:::tip + +连接器使用 Typesense 的批量导入接口。`UPDATE` 和 `DELETE` 行类型不会被解释为 CDC 操作 —— 每条上游记录都会按生成的文档 `id` 被 upsert 到目标 collection。如果希望重复作业行为类似 upsert 而不是追加，可以把 `data_save_mode` 设为 `DROP_DATA`，并配置稳定的 `primary_keys`。 + +::: + ## 任务示例 ### 使用主键写入文档 -```bash +```hocon env { parallelism = 1 job.mode = "BATCH" @@ -129,7 +143,7 @@ sink { ### 从 Typesense 读取并写入另一个 collection -```bash +```hocon env { parallelism = 1 job.mode = "BATCH" @@ -175,6 +189,47 @@ sink { } ``` +### 流式 Upsert 并按 Checkpoint 刷新 + +在流式模式下，Writer 最多缓冲 `max_batch_size` 条记录，或者直到下一个 checkpoint，再发出一次批量请求。把 `data_save_mode = DROP_DATA` 与稳定的 `primary_keys` 组合起来，每个 checkpoint 都会产生幂等的 upsert。 + +```hocon +env { + parallelism = 2 + job.mode = "STREAMING" + checkpoint.interval = 30000 +} + +source { + FakeSource { + row.num = 1000 + schema { + fields { + company_name = string + num = long + id = string + num_employees = int + flag = boolean + } + } + plugin_output = "typesense_stream" + } +} + +sink { + Typesense { + plugin_input = "typesense_stream" + hosts = ["localhost:8108"] + collection = "typesense_stream_collection" + api_key = "xyz" + primary_keys = ["id"] + max_batch_size = 100 + schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST" + data_save_mode = "DROP_DATA" + } +} +``` + ## 变更日志 diff --git a/docs/zh/connectors/source/GoogleBigtable.md b/docs/zh/connectors/source/GoogleBigtable.md index b690d4496b2f..f0eebfe9cd83 100644 --- a/docs/zh/connectors/source/GoogleBigtable.md +++ b/docs/zh/connectors/source/GoogleBigtable.md @@ -4,6 +4,10 @@ import ChangeLog from '../changelog/connector-google-bigtable.md'; > Google Bigtable Source 连接器 +## 支持这些引擎 + +> SeaTunnel Zeta
+ ## 描述 使用原生 Bigtable Data v2 Java 客户端从 Google Cloud Bigtable 读取数据。 @@ -12,12 +16,14 @@ import ChangeLog from '../changelog/connector-google-bigtable.md'; - [x] [批处理](../../introduction/concepts/connector-v2-features.md) - [ ] [流处理](../../introduction/concepts/connector-v2-features.md) -- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md) +- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) +- [ ] [列投影](../../introduction/concepts/connector-v2-features.md) - [ ] [并行度](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) :::tip -该 Source 是有界读取。当前只会为配置的表或行键范围生成一个切分，所以提高作业并行度不会把一次 Bigtable 扫描拆成多个 tablet 范围并发读取。 +该 Source 是有界读取。当前只会为配置的表或行键范围生成一个切分，所以提高作业并行度不会把一次 Bigtable 扫描拆成多个 tablet 范围并发读取。每次扫描会读取所请求行范围内的全部 Cell，并为每个 Bigtable 行输出一条 SeaTunnel 记录。 ::: @@ -52,23 +58,27 @@ Bigtable 实例 ID。 ### credentials_path [string] -Google Cloud 服务账号 JSON 密钥文件路径。未设置时使用应用默认凭证（ADC）。 +Google Cloud 服务账号 JSON 密钥文件路径。未设置时使用应用默认凭证（ADC）。在 GCE/GKE 节点上、`gcloud` shell 会话中、或当 `GOOGLE_APPLICATION_CREDENTIALS` 环境变量指向服务账号 JSON 文件时，ADC 会自动生效。 ### rowkey_column [list] -用于接收 Bigtable 行键的字段名列表。未设置时，连接器默认把名为 `rowkey` 的字段当作行键字段。行键字段可以声明为 `BYTES` 或 `STRING`。 +用于接收 Bigtable 行键的字段名列表。未设置时，连接器默认把名为 `rowkey` 的字段当作行键字段。 + +列出的每个字段会按照其在 `schema.fields` 中声明的类型独立解码：`BYTES` 接收原始行键字节；`STRING` 接收 UTF-8 解码后的视图。因此同一次扫描里的不同行键字段可以使用不同类型（例如一个字段把行键作为原始字节暴露给下游二进制处理，另一个字段同时暴露一个 UTF-8 可读视图）。 ### start_rowkey [string] 扫描起始行键，包含该行键。未设置时从表起始位置读取。 +连接器会把该值原样以 UTF-8 字符串传给 Bigtable 客户端，只支持字典序比较。对于无法按 UTF-8 编码的二进制行键，请使用 `BYTES` 类型。 + ### end_rowkey [string] 扫描结束行键，不包含该行键。未设置时读取到表末尾。 ### start_timestamp [long] -Cell 时间戳过滤的起始值，包含该时间戳，单位是微秒。 +Cell 时间戳过滤的起始值，包含该时间戳，单位是微秒。与 `end_timestamp`、`max_versions` 配合，可以控制 Bigtable 对每个列限定符返回哪些版本的 Cell。 ### end_timestamp [long] @@ -76,11 +86,11 @@ Cell 时间戳过滤的结束值，不包含该时间戳，单位是微秒。 ### max_versions [int] -每个列限定符最多返回的 Cell 版本数。默认值 `1` 表示只读取最新版本。 +每个列限定符最多返回的 Cell 版本数。默认值 `1` 表示只读取最新版本。更大的值会暴露历史版本，但 Source 仍按 Bigtable 行聚合输出，同一 Cell 的旧版本会被合并到该行返回的最新版本。 ### scan_row_limit [int] -最多读取的行数。默认值 `-1` 表示不限制。 +最多读取的行数。默认值 `-1` 表示不限制。把 `scan_row_limit` 与 `start_rowkey` / `end_rowkey` 配合，可以在多个作业之间分页扫描整张表。 ### common options @@ -102,11 +112,16 @@ Source 会读取每个 `列族:列限定符` 字段返回的最新 Cell。可以 ::: -## 示例 +## 任务示例 ### 使用应用默认凭证读取整张表 ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + source { GoogleBigtable { project_id = "my-gcp-project" @@ -126,6 +141,11 @@ source { ### 使用服务账号扫描行键范围 ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + source { GoogleBigtable { project_id = "my-gcp-project" @@ -149,6 +169,11 @@ source { ### 使用自定义行键字段名 ```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + source { GoogleBigtable { project_id = "my-gcp-project" @@ -166,6 +191,38 @@ source { } ``` +### 有界流式扫描并按 Cell 版本过滤 + +在 `STREAMING` 模式下，仍然只做单次有界扫描，但会按 checkpoint 推进。结合 `start_timestamp`、`end_timestamp` 和 `max_versions` 可以限制 Bigtable 返回的 Cell 版本。 + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 60000 +} + +source { + GoogleBigtable { + project_id = "my-gcp-project" + instance_id = "my-bigtable-instance" + table = "events" + start_timestamp = 1704067200000000 + end_timestamp = 1735689600000000 + max_versions = 3 + scan_row_limit = 500000 + schema { + fields { + rowkey = STRING + "cf:type" = STRING + "cf:data" = STRING + "cf:ts" = BIGINT + } + } + } +} +``` + ## Changelog diff --git a/docs/zh/connectors/source/Typesense.md b/docs/zh/connectors/source/Typesense.md index 40937446ceab..dd412d3a46e5 100644 --- a/docs/zh/connectors/source/Typesense.md +++ b/docs/zh/connectors/source/Typesense.md @@ -4,16 +4,23 @@ import ChangeLog from '../changelog/connector-typesense.md'; > Typesense 源连接器 +## 支持的引擎 + +> SeaTunnel Zeta
+ ## 描述 从 Typesense collection 读取文档。该 source 支持有界批读取，也可以通过 `query` 传入 Typesense 查询参数。 +Source 是有界读取：每次作业只会读取一次匹配 `query` 的全部文档，然后结束。如果需要变更捕获，请在接收端借助 Typesense 自身的变化追踪机制实现。 + ## 主要特性 - [x] [批处理](../../introduction/concepts/connector-v2-features.md) - [ ] [流处理](../../introduction/concepts/connector-v2-features.md) - [ ] [精确一次](../../introduction/concepts/connector-v2-features.md) +- [ ] [cdc](../../introduction/concepts/connector-v2-features.md) - [x] [Schema](../../introduction/concepts/connector-v2-features.md) - [x] [并行度](../../introduction/concepts/connector-v2-features.md) - [ ] [支持用户定义的拆分](../../introduction/concepts/connector-v2-features.md) @@ -33,7 +40,7 @@ Typesense 查询参数。 ### hosts [array] -Typesense 的访问地址，格式为 `host:port`，例如：`["typesense-01:8108"]`。支持配置多个地址。 +Typesense 的访问地址，格式为 `host:port`，例如：`["typesense-01:8108"]`。支持配置多个地址。配置多个节点时，Source 会把搜索请求发往第一个可达节点，列表本身不用于并行扫描分片。 ### collection [string] @@ -45,7 +52,7 @@ typesense 需要读取的列。有关更多信息，请参阅：[guide](../../in ### api_key [string] -Typesense 安全认证的 `api_key`。 +Typesense 安全认证的 `api_key`。请把它当作敏感凭据处理；在共享环境运行时，建议通过作业密钥或环境变量注入。 ### protocol [string] @@ -56,9 +63,11 @@ Typesense 安全认证的 `api_key`。 Typesense 查询参数，例如 `q=*&filter_by=num_employees:>9000`。不配置时读取默认查询返回的文档。 +所有合法的 Typesense 搜索参数都可以追加进来，包括 `q`、`query_by`、`filter_by`、`sort_by`、`page` 和 `per_page`，连接器会原样转发给 Typesense 搜索接口。 + ### batch_size [int] -读取数据时每批查询的文档数量。 +读取数据时每批查询的文档数量。每次请求使用 Typesense 的 `per_page` 参数，因此该值必须在 1 与 Typesense 服务端 `per_page` 上限（通常是 250）之间。如果日志中看到分页被截断，请调小该值。 ### 常用选项 @@ -68,7 +77,7 @@ Source 插件常用参数，具体请参考 [Source 常用选项](../common-opti ### 带过滤条件读取文档 -```bash +```hocon env { parallelism = 1 job.mode = "BATCH" @@ -103,6 +112,39 @@ sink { } ``` +### 使用自定义查询条件读取子集 + +组合 `query_by` 与 `sort_by` 可以控制 Typesense 搜索的字段以及结果集的排序方式。 + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + Typesense { + hosts = ["localhost:8108"] + collection = "companies" + api_key = "xyz" + query = "q=acme&query_by=company_name&filter_by=country:=US&sort_by=num_employees:desc" + batch_size = 50 + schema = { + fields { + company_name = string + num_employees = long + country = string + id = string + } + } + } +} + +sink { + Console {} +} +``` + ## 变更日志 diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java index 90dbe021fc62..b837827e5d10 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java @@ -38,6 +38,7 @@ import java.io.IOException; import java.io.Serializable; +import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; @@ -57,6 +58,7 @@ public class JdbcInputFormat implements Serializable { private final JdbcRowConverter jdbcRowConverter; private final Map tables; private final ChunkSplitter chunkSplitter; + private final boolean configuredAutoCommit; private transient String splitTableId; private transient TableSchema splitTableSchema; @@ -65,23 +67,37 @@ public class JdbcInputFormat implements Serializable { private volatile boolean hasNext; public JdbcInputFormat(JdbcSourceConfig config, Map tables) { - this.jdbcDialect = + this( JdbcDialectLoader.load( config.getJdbcConnectionConfig().getUrl(), config.getJdbcConnectionConfig().getDialect(), - config.getCompatibleMode()); - this.chunkSplitter = ChunkSplitter.create(config); + config.getCompatibleMode()), + ChunkSplitter.create(config), + tables, + config.getJdbcConnectionConfig().isAutoCommit()); + } + + JdbcInputFormat( + JdbcDialect jdbcDialect, + ChunkSplitter chunkSplitter, + Map tables, + boolean configuredAutoCommit) { + this.jdbcDialect = jdbcDialect; + this.chunkSplitter = chunkSplitter; this.jdbcRowConverter = jdbcDialect.getRowConverter(); this.tables = tables; + this.configuredAutoCommit = configuredAutoCommit; } public void openInputFormat() {} public void closeInputFormat() throws IOException { - close(); - - if (chunkSplitter != null) { - chunkSplitter.close(); + try { + close(); + } finally { + if (chunkSplitter != null) { + chunkSplitter.close(); + } } } @@ -101,10 +117,31 @@ public void open(JdbcSourceSplit inputSplit) throws IOException { resultSet = statement.executeQuery(); hasNext = resultSet.next(); } catch (SQLException se) { + cleanupAfterOpenFailure(se); throw new JdbcConnectorException( JdbcConnectorErrorCode.CONNECT_DATABASE_FAILED, "open() failed." + se.getMessage(), se); + } catch (RuntimeException runtimeException) { + cleanupAfterOpenFailure(runtimeException); + throw runtimeException; + } + } + + private void cleanupAfterOpenFailure(Throwable openException) { + boolean shouldDiscardConnection = statement == null; + try { + close(); + } catch (IOException cleanupException) { + openException.addSuppressed(cleanupException); + shouldDiscardConnection = true; + } finally { + if (shouldDiscardConnection) { + // Statement creation may establish or mutate the cached connection before failing + // without returning a statement. Discard it because close() cannot identify and + // finish that transaction safely. + chunkSplitter.close(); + } } } @@ -114,11 +151,14 @@ public void open(JdbcSourceSplit inputSplit) throws IOException { * @throws IOException Indicates that a resource could not be closed. */ public void close() throws IOException { + Connection connection = getStatementConnection(); if (resultSet != null) { try { resultSet.close(); } catch (SQLException e) { LOG.info("ResultSet couldn't be closed - " + e.getMessage()); + } finally { + resultSet = null; } } if (statement != null) { @@ -126,7 +166,86 @@ public void close() throws IOException { statement.close(); } catch (SQLException e) { LOG.info("Statement couldn't be closed - " + e.getMessage()); + } finally { + statement = null; + } + } + + hasNext = false; + splitTableSchema = null; + splitTableId = null; + finishReadTransaction(connection); + } + + private Connection getStatementConnection() { + if (statement == null) { + return null; + } + try { + Connection connection = statement.getConnection(); + if (connection == null) { + LOG.warn( + "The JDBC source statement returned no connection. " + + "Closing the cached connection to avoid reusing an unknown " + + "transaction."); + chunkSplitter.close(); } + return connection; + } catch (SQLException e) { + LOG.warn( + "Failed to get the JDBC source connection from the current statement. " + + "Closing the cached connection to avoid reusing an unknown " + + "transaction.", + e); + chunkSplitter.close(); + return null; + } + } + + private void finishReadTransaction(Connection connection) { + try { + finishReadTransaction(connection, configuredAutoCommit); + } catch (SQLException e) { + LOG.warn( + "Failed to finish the JDBC source read transaction. " + + "Closing the connection to avoid leaving or reusing an idle " + + "transaction.", + e); + discardConnection(connection, e); + } + } + + private void discardConnection(Connection connection, SQLException cleanupException) { + try { + if (connection != null) { + connection.close(); + } + } catch (SQLException closeException) { + cleanupException.addSuppressed(closeException); + LOG.warn( + "Failed to close the JDBC source connection after transaction cleanup failed.", + cleanupException); + } finally { + // Clear the provider's cached reference and retry close for drivers whose first close + // attempt failed. + chunkSplitter.close(); + } + } + + static void finishReadTransaction(Connection connection, boolean configuredAutoCommit) + throws SQLException { + if (connection == null || connection.isClosed()) { + return; + } + + boolean currentAutoCommit = connection.getAutoCommit(); + if (!currentAutoCommit) { + // JDBC source reads do not have changes to commit. Rollback ends the server-side cursor + // transaction, releases its snapshot, and also recovers a transaction in failed state. + connection.rollback(); + } + if (currentAutoCommit != configuredAutoCommit) { + connection.setAutoCommit(configuredAutoCommit); } } diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormatTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormatTest.java new file mode 100644 index 000000000000..13f5fb72f5e7 --- /dev/null +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormatTest.java @@ -0,0 +1,272 @@ +/* + * 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.jdbc.internal; + +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException; +import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter; +import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect; +import org.apache.seatunnel.connectors.seatunnel.jdbc.source.ChunkSplitter; +import org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceSplit; + +import org.junit.jupiter.api.Test; +import org.mockito.InOrder; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.Collections; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class JdbcInputFormatTest { + + private static final TablePath TABLE_PATH = TablePath.of("test", "public", "source_table"); + private static final TableSchema TABLE_SCHEMA = TableSchema.builder().build(); + private static final JdbcSourceSplit SPLIT = + new JdbcSourceSplit(TABLE_PATH, "split-0", null, null, null, null, null); + + @Test + void shouldRollbackAndRestoreConfiguredAutoCommitAfterClosingSplit() throws Exception { + TestContext context = createContext(true); + context.openEmptySplit(); + when(context.connection.isClosed()).thenReturn(false); + when(context.connection.getAutoCommit()).thenReturn(false); + + context.inputFormat.close(); + + InOrder inOrder = inOrder(context.statement, context.resultSet, context.connection); + inOrder.verify(context.statement).getConnection(); + inOrder.verify(context.resultSet).close(); + inOrder.verify(context.statement).close(); + inOrder.verify(context.connection).rollback(); + inOrder.verify(context.connection).setAutoCommit(true); + + context.inputFormat.close(); + verify(context.connection, times(1)).rollback(); + } + + @Test + void shouldRollbackAndKeepConfiguredManualCommit() throws Exception { + TestContext context = createContext(false); + context.openEmptySplit(); + when(context.connection.isClosed()).thenReturn(false); + when(context.connection.getAutoCommit()).thenReturn(false); + + context.inputFormat.close(); + + verify(context.connection).rollback(); + verify(context.connection, never()).setAutoCommit(true); + } + + @Test + void shouldNotRollbackAutoCommitConnection() throws Exception { + TestContext context = createContext(true); + context.openEmptySplit(); + when(context.connection.isClosed()).thenReturn(false); + when(context.connection.getAutoCommit()).thenReturn(true); + + context.inputFormat.close(); + + verify(context.connection, never()).rollback(); + verify(context.connection, never()).setAutoCommit(true); + } + + @Test + void shouldFinishTransactionWhenResourceCloseFails() throws Exception { + TestContext context = createContext(true); + context.openEmptySplit(); + when(context.connection.isClosed()).thenReturn(false); + when(context.connection.getAutoCommit()).thenReturn(false); + doThrow(new SQLException("result set close failed")).when(context.resultSet).close(); + doThrow(new SQLException("statement close failed")).when(context.statement).close(); + + context.inputFormat.close(); + + verify(context.connection).rollback(); + verify(context.connection).setAutoCommit(true); + } + + @Test + void shouldKeepConnectionReusableAcrossSuccessfulSplits() throws Exception { + TestContext context = createContext(true); + PreparedStatement secondStatement = mock(PreparedStatement.class); + ResultSet secondResultSet = mock(ResultSet.class); + + when(context.chunkSplitter.generateSplitStatement(SPLIT, TABLE_SCHEMA)) + .thenReturn(context.statement, secondStatement); + when(context.statement.executeQuery()).thenReturn(context.resultSet); + when(secondStatement.executeQuery()).thenReturn(secondResultSet); + when(context.resultSet.next()).thenReturn(false); + when(secondResultSet.next()).thenReturn(false); + when(context.statement.getConnection()).thenReturn(context.connection); + when(secondStatement.getConnection()).thenReturn(context.connection); + when(context.connection.isClosed()).thenReturn(false); + when(context.connection.getAutoCommit()).thenReturn(false); + + context.inputFormat.open(SPLIT); + context.inputFormat.close(); + context.inputFormat.open(SPLIT); + context.inputFormat.close(); + + verify(context.chunkSplitter, times(2)).generateSplitStatement(SPLIT, TABLE_SCHEMA); + verify(context.connection, times(2)).rollback(); + verify(context.connection, times(2)).setAutoCommit(true); + verify(context.connection, never()).close(); + verify(context.chunkSplitter, never()).close(); + } + + @Test + void shouldDiscardConnectionWhenTransactionCleanupFails() throws Exception { + TestContext context = createContext(true); + context.openEmptySplit(); + when(context.connection.isClosed()).thenReturn(false); + when(context.connection.getAutoCommit()).thenReturn(false); + doThrow(new SQLException("rollback failed")).when(context.connection).rollback(); + + context.inputFormat.close(); + + verify(context.connection).close(); + verify(context.chunkSplitter).close(); + verify(context.connection, never()).setAutoCommit(true); + } + + @Test + void shouldCloseCachedConnectionWhenStatementCannotExposeConnection() throws Exception { + TestContext context = createContext(true); + context.openEmptySplit(); + when(context.statement.getConnection()) + .thenThrow(new SQLException("get connection failed")); + + context.inputFormat.close(); + + verify(context.chunkSplitter).close(); + verify(context.resultSet).close(); + verify(context.statement).close(); + } + + @Test + void shouldDiscardCachedConnectionWhenStatementCreationFails() throws Exception { + TestContext context = createContext(true); + when(context.chunkSplitter.generateSplitStatement(SPLIT, TABLE_SCHEMA)) + .thenThrow(new SQLException("prepare failed")); + + assertThrows(JdbcConnectorException.class, () -> context.inputFormat.open(SPLIT)); + + verify(context.chunkSplitter).close(); + verify(context.statement, never()).executeQuery(); + } + + @Test + void shouldRollbackAndKeepConnectionWhenExecuteQueryFails() throws Exception { + TestContext context = createContext(true); + when(context.chunkSplitter.generateSplitStatement(SPLIT, TABLE_SCHEMA)) + .thenReturn(context.statement); + when(context.statement.getConnection()).thenReturn(context.connection); + when(context.statement.executeQuery()).thenThrow(new SQLException("execute failed")); + when(context.connection.isClosed()).thenReturn(false); + when(context.connection.getAutoCommit()).thenReturn(false); + + assertThrows(JdbcConnectorException.class, () -> context.inputFormat.open(SPLIT)); + + verify(context.statement).close(); + verify(context.connection).rollback(); + verify(context.connection).setAutoCommit(true); + verify(context.connection, never()).close(); + verify(context.chunkSplitter, never()).close(); + } + + @Test + void shouldAlwaysCloseCachedConnectionWhenInputFormatCloses() throws Exception { + TestContext context = createContext(true); + + context.inputFormat.closeInputFormat(); + + verify(context.chunkSplitter).close(); + } + + @Test + void shouldIgnoreClosedConnection() throws SQLException { + Connection connection = mock(Connection.class); + when(connection.isClosed()).thenReturn(true); + + JdbcInputFormat.finishReadTransaction(connection, true); + + verify(connection, never()).getAutoCommit(); + verify(connection, never()).rollback(); + } + + private static TestContext createContext(boolean configuredAutoCommit) throws SQLException { + JdbcDialect dialect = mock(JdbcDialect.class); + JdbcRowConverter rowConverter = mock(JdbcRowConverter.class); + ChunkSplitter chunkSplitter = mock(ChunkSplitter.class); + PreparedStatement statement = mock(PreparedStatement.class); + ResultSet resultSet = mock(ResultSet.class); + Connection connection = mock(Connection.class); + CatalogTable catalogTable = mock(CatalogTable.class); + Map tables = Collections.singletonMap(TABLE_PATH, catalogTable); + + when(dialect.getRowConverter()).thenReturn(rowConverter); + when(catalogTable.getTableSchema()).thenReturn(TABLE_SCHEMA); + + JdbcInputFormat inputFormat = + new JdbcInputFormat(dialect, chunkSplitter, tables, configuredAutoCommit); + return new TestContext(inputFormat, chunkSplitter, statement, resultSet, connection); + } + + private static final class TestContext { + private final JdbcInputFormat inputFormat; + private final ChunkSplitter chunkSplitter; + private final PreparedStatement statement; + private final ResultSet resultSet; + private final Connection connection; + + private TestContext( + JdbcInputFormat inputFormat, + ChunkSplitter chunkSplitter, + PreparedStatement statement, + ResultSet resultSet, + Connection connection) { + this.inputFormat = inputFormat; + this.chunkSplitter = chunkSplitter; + this.statement = statement; + this.resultSet = resultSet; + this.connection = connection; + } + + private void openEmptySplit() throws IOException, SQLException { + when(chunkSplitter.generateSplitStatement(SPLIT, TABLE_SCHEMA)).thenReturn(statement); + when(statement.executeQuery()).thenReturn(resultSet); + when(resultSet.next()).thenReturn(false); + when(statement.getConnection()).thenReturn(connection); + inputFormat.open(SPLIT); + } + } +}