[ISSUE-817] Add geaflow-dsl-connector-doris (Stream Load sink + partitioned source)#819
[ISSUE-817] Add geaflow-dsl-connector-doris (Stream Load sink + partitioned source)#819E2ern1ty wants to merge 6 commits into
Conversation
…tioned source) - Sink writes to Doris via Stream Load: buffer rows in write(), flush a whole batch on row/byte threshold or window finish(), giving much higher throughput than row-by-row JDBC INSERT. CSV payload by default, JSON supported. - Source reads through the MySQL protocol with parallel partitioned reads. - SPI registration + parent/aggregation pom wiring + runtime dependency. - CN + EN docs and examples. - Unit tests and a Testcontainers integration test with a throughput benchmark (self-skips when Docker/host-networking is unavailable).
The Doris Testcontainers integration test previously ran whenever Docker was available on a Linux host, which is exactly the case on GitHub CI runners, so it would try to pull the multi-GB Doris image and start a container during a normal 'mvn test'. Make it opt-in: it only runs with -Ddoris.it.enabled=true or when an external Doris is provided via -Ddoris.it.fenodes/-Ddoris.it.jdbcUrl, and otherwise skips. Docs updated accordingly.
…E failover Address the maintainer's review points on the Doris connector: - serialization: default to json (Gson with serializeNulls + disableHtmlEscaping) so newlines, quotes, backslashes, unicode, null and empty strings round-trip correctly; document csv as a no-escape plain-split option. - sink flush: flushBuffer stays safe on an empty buffer and is idempotent on finish(); close() now flushes the remaining buffer and is safe on repeated calls and when the client is not initialized. - partition logic: clamp partition.num to at least 1 instead of throwing; the first partition already covers rows whose partition column is entirely NULL. - label: kept globally unique via UUID. - FE failover: DorisStreamLoad now takes the full FE list and rotates to the next FE on a failed request (retries at least once per FE). - integration tests: add table-not-found, FE-failover and full data-type round-trip fidelity cases (still opt-in). Add unit tests for special-char / null / empty-string serialization and multi-FE url building.
- Make mysql-connector-java a test-scope dependency instead of compile, so the GPL-licensed driver is not shipped in the distribution (matching geaflow-dsl-connector-jdbc, which bundles no driver). The source only uses the standard java.sql API and loads the driver by name, so users must provide a MySQL-protocol driver on the runtime classpath; documented in CN + EN docs. - Remove the unused COLON constant. Verified: checkstyle 0 violations, RAT 0 unapproved (14 files), unit tests 23 run / 0 failed / 7 skipped (integration tests are opt-in).
Verified the connector end to end against a real single-node apache/doris:3.0.8-all: the sink (Stream Load), the partitioned source, FE failover, table-not-found and full data-type round-trip (newlines, quotes, backslashes, chinese, emoji) integration tests all pass. Benchmark: Stream Load loaded 500 rows in ~91ms (~5500 rows/s) vs ~11335ms (~44 rows/s) for row-by-row JDBC INSERT, ~125x faster; 2000 rows via Stream Load took only ~97ms. Recorded the reference numbers in the CN + EN docs.
| } | ||
|
|
||
| @Override | ||
| public List<Partition> listPartitions() { |
There was a problem hiding this comment.
Problem/Cause: Currently, only numeric range partitioning is supported. When columns are highly skewed, have no indexes, or are non-numeric (string/date), the partitioning strategy may lead to hotspots or uneven partitioning.
Recommendation: Provide multiple partitioning strategies (range, hash, sampling-based), allow users to manually specify partition boundaries, or use custom SQL subqueries. Emphasize the performance impact of unindexed partitioning columns in the documentation and recommend indexing/appropriate partitioning columns.
There was a problem hiding this comment.
Thanks. Added a source.partition.mode option (range/custom). custom takes semicolon-separated WHERE predicates via source.partition.clauses (one partition per predicate), which supports non-numeric/skewed columns and arbitrary conditions (e.g. date subranges), so users can specify partition boundaries manually. Kept numeric range as the default. Docs now warn that the range partition column should be indexed and evenly distributed (otherwise full scans/hotspots) and recommend the primary key. Added DorisTableSourceTest covering single/range/custom listing. Sampling-based auto splitting can be a follow-up.
| } | ||
| } | ||
|
|
||
| private String generateLabel() { |
There was a problem hiding this comment.
Partial Success and Duplicate Writes (Label/Idempotency)
Problem/Cause: Stream Load uses labels/UUIDs for assistance, but after FE failover and retries, there is a risk of duplicate writes or semantic inconsistencies due to "partial success + retries" (Doris's behavior for the same label needs to be explicitly defined).
Recommendation: Clearly define the semantics of labels in the README and implementation (if the server supports idempotency/duplicate checks), and:
Log the label in the failure/retry strategy and check the status of the label on the server before retrying (if it is queryable).
Provide optional "idempotency/duplicate removal strategies" (e.g., allow-duplicates vs. check-label), or expose the number of retries and their behavior as configuration options.
There was a problem hiding this comment.
Fixed the idempotency risk: the label is now generated once per batch and reused across every retry and FE failover attempt (previously a new label was created per attempt, which defeated Doris's dedup). Since Doris loads a given label at most once, a retry after a mistaken failure now hits "Label Already Exists", which is treated as success instead of writing the batch twice. The label is logged on every failure/retry. Documented these write semantics in the CN/EN docs. Retry behavior is configurable via sink.max.retries (kept at least one attempt per FE for failover).
| // Rotate over the FE list so a failed request fails over to the next FE. | ||
| String url = loadUrls.get(attempt % loadUrls.size()); | ||
| try { | ||
| doLoad(url, payload); |
There was a problem hiding this comment.
Connection/Read/Write Timeout and Large Payload Timeout
Problem/Cause: Stream Load request bodies can be very large. A default timeout that is too short can cause interruption; a timeout that is too long can easily exhaust resources in the event of network anomalies.
Recommendation: Set reasonable connection and read/write timeouts, and allow users to configure them. Provide chunked/streaming uploads for large requests (if Doris supports this) or add upload timeout instructions.
There was a problem hiding this comment.
Connect and read/write timeouts are both user-configurable (request.connect.timeout.ms, request.read.timeout.ms) and applied to the HttpClient RequestConfig. Raised the default read/write timeout to 60s so it covers uploading a large batch, and documented tuning guidance (increase for very large batches on slow networks, lower to fail faster on anomalies). The payload is already sent as a single streamed HTTP entity; batch size itself is bounded by sink.batch.rows/sink.batch.bytes so a batch stays within a reasonable size.
|
Hello @E2ern1ty, I'm glad you contributed to the community! I left some comments, please take a look and see if they need any changes. |
…, timeouts - Label/idempotency: generate one label per batch and reuse it across all retries and FE failover attempts (Doris loads a label at most once); treat 'Label Already Exists' on retry as success to avoid duplicate writes; log the label on failures/retries. Documented the write semantics. - Partition strategies: add source.partition.mode (range/custom). 'custom' takes semicolon-separated WHERE predicates (source.partition.clauses), one partition per predicate, supporting non-numeric/skewed columns and arbitrary conditions. Documented the performance impact of unindexed/skewed partition columns and recommended indexed, evenly-distributed columns. - Timeouts: read/write timeout is configurable and its default is raised to 60s to cover large-batch uploads; documented tuning guidance. - Tests: add DorisTableSourceTest covering single/range/custom partition listing.
2 participants
What changes were proposed in this pull request?
Closes #817.
Add a new
geaflow-dsl-connector-dorisconnector so graph-computing results can be written to / read from Apache Doris efficiently. Doris speaks the MySQL protocol so it is reachable today via the JDBC connector, but JDBC only does row-by-row INSERT and misses Doris's high-throughput load path — Stream Load. This connector adds it.Sink (Stream Load)
write()and flushes a whole batch via a single HTTP PUT to the FE/_stream_loadendpoint once the row-count or byte-size threshold is reached, or on windowfinish().Status.serializeNulls+disableHtmlEscaping), which correctly handles newlines, quotes, backslashes, unicode, null and empty strings; CSV is available as a no-escape high-throughput option.Source (partitioned reads)
geaflow-dsl-connector-jdbc); users provide a MySQL-protocol driver on the runtime classpath.Also: SPI registration, parent/aggregation pom + runtime wiring, and CN + EN docs with options, examples and benchmark numbers.
Addresses the maintainer's review points: safe CSV/JSON serialization; idempotent
finish()and empty-buffer-safe flush;close()flushes the remaining buffer and is safe on repeated calls;partition.numis clamped; globally-unique labels via UUID; FE failover; and integration tests covering FE failover, table-not-found and full data-type round-trip fidelity.How was this PR tested?
Tests have Added for the changes
Production environment verified
Unit tests (run in CI): serialization of special chars / null / empty string, config defaults, multi-FE url building, sink init validation.
checkstyle0 violations,RAT0 unapproved.Testcontainers integration test (opt-in, disabled by default so CI stays green): sink+source round-trip, FE failover, table-not-found, full data-type fidelity, and a throughput benchmark.
Verified end to end against a real single-node
apache/doris:3.0.8-all: all four core integration cases pass. Benchmark: Stream Load loaded 500 rows in ~91ms (~5500 rows/s) vs ~11335ms (~44 rows/s) for row-by-row JDBC INSERT (~125x faster); 2000 rows via Stream Load took only ~97ms.