feat(sql): add pluggable source/sink builders and sink passthrough connectors

Refactor connector config construction into role-specific providers and builders, and add sink passthrough implementations for filesystem, delta, iceberg, lancedb plus dedicated S3/Kafka provider wiring for stronger option validation.

Made-with: Cursor

Author: luoluoyuyu

Cargo.lock

@@ -52,6 +52,8 @@ arrow-array = "55"
 arrow-ipc = "55"
 arrow-schema = { version = "55", features = ["serde"] }
 parquet = "55"
+object_store = { version = "0.12.5", features = ["aws"] }
+bytes = "1"
 futures = "0.3"
 serde_json_path = "0.7"
 xxhash-rust = { version = "0.8", features = ["xxh3"] }
@@ -63,6 +65,7 @@ itertools = "0.14"
 strum = { version = "0.26", features = ["derive"] }
 
 arrow-json = {version = '55.2.0'}
+apache-avro = "0.21"
 datafusion = {git = 'https://github.com/FunctionStream/datafusion', branch = '48.0.1/fs'}
 datafusion-common = {git = 'https://github.com/FunctionStream/datafusion', branch = '48.0.1/fs'}
 datafusion-execution = {git = 'https://github.com/FunctionStream/datafusion', branch = '48.0.1/fs'}
@@ -74,6 +77,9 @@ sqlparser = { git = "https://github.com/FunctionStream/sqlparser-rs", branch = "
 
 ahash = "0.8"
 governor = "0.8.0"
+lance = { version = "4.0.0", default-features = false, features = ["aws"] }
+arrow-array-lance = { package = "arrow-array", version = "57.3.0" }
+arrow-ipc-lance = { package = "arrow-ipc", version = "57.3.0" }
 
 [features]
 default = ["incremental-cache", "python"]

Cargo.toml

@@ -9,8 +9,6 @@
 # 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.
-
-
 APP_NAME    := function-stream
 VERSION     := $(shell grep '^version' Cargo.toml | head -1 | awk -F '"' '{print $$2}')
 DATE        := $(shell date -u +"%Y-%m-%dT%H:%M:%SZ")
@@ -29,30 +27,22 @@ endif
 OS          := $(shell uname -s | tr '[:upper:]' '[:lower:]')
 OS_NAME     := $(shell uname -s)
 
-# 2. Configure RUSTFLAGS and target triple per platform
+# 2. Configure target triple per platform
 DIST_ROOT   := dist
 ifeq ($(OS_NAME), Linux)
     TRIPLE := $(ARCH)-unknown-linux-gnu
     STATIC_FLAGS :=
 else ifeq ($(OS_NAME), Darwin)
-    # macOS: strip symbols but keep dynamic linking (Apple system restriction)
     TRIPLE := $(ARCH)-apple-darwin
     STATIC_FLAGS :=
 else ifneq (,$(findstring MINGW,$(OS_NAME))$(findstring MSYS,$(OS_NAME)))
-    # Windows (Git Bash / MSYS2): static-link MSVC runtime
     TRIPLE := $(ARCH)-pc-windows-msvc
     STATIC_FLAGS := -C target-feature=+crt-static
 else
-    # Fallback
     TRIPLE := $(ARCH)-unknown-linux-gnu
     STATIC_FLAGS :=
 endif
 
-# 3. Aggressive optimization flags
-# opt-level=z  : size-oriented, minimize binary footprint
-# strip=symbols: remove debug symbol table at link time
-# Note: panic=abort is intentionally omitted to preserve stack unwinding
-#       for better fault tolerance in the streaming runtime
 OPTIMIZE_FLAGS := -C opt-level=z -C strip=symbols $(STATIC_FLAGS)
 
 TARGET_DIR  := target/$(TRIPLE)/release
@@ -110,7 +100,7 @@ help:
 	(printf "$(C_Y)[!] Auto-installing target toolchain for $(OS_NAME): $(TRIPLE)$(C_0)\n" && \
 	 rustup target add $(TRIPLE))
 
-# 5. Build targets (depend on .ensure-target for automatic toolchain setup)
+# 5. Build targets
 build: .check-env .ensure-target .build-wasm
 	$(call log,BUILD,Rust Full [$(OS_NAME) / $(TRIPLE)])
 	@RUSTFLAGS="$(OPTIMIZE_FLAGS)" \
@@ -128,24 +118,24 @@ build: .check-env .ensure-target .build-wasm
 
 build-lite: .check-env .ensure-target
 	$(call log,BUILD,Rust Lite [$(OS_NAME) / $(TRIPLE)])
-	@RUSTFLAGS="$(OPTIMIZE_FLAGS)" \
+	@RUSTFLAGS="$(INDUSTRIAL_RUSTFLAGS)" \
 	cargo build --release \
-		--target $(TRIPLE) \
-		--no-default-features \
-		--features incremental-cache \
-		--quiet
+	   --target $(TRIPLE) \
+	   --no-default-features \
+	   --features incremental-cache \
+	   --quiet
 	$(call log,BUILD,CLI for dist)
-	@RUSTFLAGS="$(OPTIMIZE_FLAGS)" \
+	@RUSTFLAGS="$(INDUSTRIAL_RUSTFLAGS)" \
 	cargo build --release \
-		--target $(TRIPLE) \
-		-p function-stream-cli \
-		--quiet
+	   --target $(TRIPLE) \
+	   -p function-stream-cli \
+	   --quiet
 	$(call success,Target: $(TARGET_DIR)/$(APP_NAME) $(TARGET_DIR)/cli)
 
 .build-wasm:
 	$(call log,WASM,Building Python Runtime using $(PYTHON_EXEC))
 	@cd $(PYTHON_ROOT)/functionstream-runtime && \
-		PYTHONPATH=../functionstream-api:../functionstream-api-advanced ../../$(PYTHON_EXEC) build.py > /dev/null
+	   PYTHONPATH=../functionstream-api:../functionstream-api-advanced ../../$(PYTHON_EXEC) build.py > /dev/null
 	@[ -f "$(WASM_SOURCE)" ] || (printf "$(C_R)[X] WASM Build Failed$(C_0)\n" && exit 1)
 
 dist: build
@@ -223,9 +213,9 @@ docker:
 docker-run:
 	$(call log,DOCKER,Starting Container)
 	@docker run --rm -it \
-		-p 8080:8080 \
-		-v $(CURDIR)/logs:/app/logs \
-		$(IMAGE_NAME)
+	   -p 8080:8080 \
+	   -v $(CURDIR)/logs:/app/logs \
+	   $(IMAGE_NAME)
 
 docker-push:
 	$(call log,DOCKER,Pushing $(IMAGE_NAME))

Makefile

@@ -204,6 +204,7 @@ function-stream-<version>/
 | 文档                                                                     | 描述                       |
 |------------------------------------------------------------------------|--------------------------|
 | [Streaming SQL 使用指南](docs/streaming-sql-guide-zh.md)                   | 声明式 SQL 实时流处理指南         |
+| [Streaming SQL Source/Sink 文档](docs/streaming-sql/README-zh.md)          | Source/Sink 能力与参数速查         |
 | [连接器、格式与类型参考](docs/connectors-and-formats-zh.md)                       | 支持的 Source/Sink、格式与数据类型  |
 | [服务端配置与运维指南](docs/server-configuration-zh.md)                          | 服务端配置与运维操作              |
 | [Function 任务配置规范](docs/function-configuration-zh.md)        | 任务定义规范           |

README-zh.md

@@ -90,6 +90,8 @@ flowchart LR
 | **水位线 (Watermark)** | `AS <column> - INTERVAL ...` | 对**迟到、乱序**数据的容忍度；时间推进由水位线驱动，**过度迟到**的事件会被安全丢弃。 |
 
 > **完整参考**：支持的连接器、数据格式和 SQL 数据类型，请参阅 [连接器、格式与类型参考](connectors-and-formats-zh.md)。
+>
+> **Source / Sink 专项说明**：请参阅 [Streaming SQL Connector 文档](streaming-sql/README-zh.md)（含 Kafka Source 与 filesystem/s3/delta/iceberg/lanceDB Sink）。
 
 ---
 

docs/streaming-sql-guide-zh.md

@@ -0,0 +1,13 @@
+# Streaming SQL Connector 文档
+
+本目录提供 Streaming SQL 的 Source / Sink 专项文档，推荐配合 `CREATE STREAMING TABLE ... AS SELECT ...` 使用。
+
+## 目录
+
+- [Source 文档](Source/README-zh.md)
+- [Sink 文档](Sink/README-zh.md)
+
+## 使用建议
+
+1. 先用 `CREATE TABLE ... WITH (...)` 注册 Source（当前仅 Kafka）。
+2. 再用 `CREATE STREAMING TABLE ... WITH (...) AS SELECT ...` 创建持续运行的 Pipeline 并写入 Sink。

docs/streaming-sql/README-zh.md

@@ -0,0 +1,14 @@
+# Streaming SQL Connector Docs
+
+This directory contains Source/Sink focused docs for Streaming SQL, intended to be used with `CREATE STREAMING TABLE ... AS SELECT ...`.
+
+## Index
+
+- [Source Docs](Source/README.md)
+- [Sink Docs](Sink/README.md)
+
+## Recommended workflow
+
+1. Register sources using `CREATE TABLE ... WITH (...)` (currently Kafka source).
+2. Build a continuous pipeline using `CREATE STREAMING TABLE ... WITH (...) AS SELECT ...` and write to sinks.
+

docs/streaming-sql/README.md

@@ -0,0 +1,31 @@
+# Streaming SQL Sink 文档
+
+本目录聚焦 Streaming SQL 的下游写出能力（Sink）。
+
+## 支持矩阵
+
+| Connector | 支持格式 |
+
+|---|---|
+| `kafka` | `json` / `raw_string` / `raw_bytes`（沿用 Kafka Sink 编码能力） |
+| `filesystem` | `csv` / `parquet` / `json`(JSONL) / `avro` / `orc` |
+| `s3` | `csv` / `parquet` |
+| `delta` | `csv` / `parquet` / `json`(JSONL) / `avro` / `orc` |
+| `iceberg` | `csv` / `parquet` |
+| `lanceDB` | `lance` |
+
+## 文档列表
+
+- [Kafka Sink](kafka-sink-zh.md)
+- [Filesystem Sink](filesystem-sink-zh.md)
+- [S3 Sink](s3-sink-zh.md)
+- [Delta Sink](delta-sink-zh.md)
+- [Iceberg Sink](iceberg-sink-zh.md)
+- [LanceDB Sink](lancedb-sink-zh.md)
+
+## 通用约定
+
+- 在 `CREATE STREAMING TABLE ... WITH (...) AS SELECT ...` 中通过 `WITH` 指定 `connector` 与 `format`。
+- Sink 场景建议显式指定 `type='sink'`。
+- 仅 `lanceDB` connector 允许 `format='lance'`；其余 Sink connector 不支持 `lance`。
+- `format='json'` 的文件类 Sink 输出为 JSON Lines（NDJSON，`.jsonl`）。

docs/streaming-sql/Sink/README-zh.md

@@ -0,0 +1,31 @@
+# Streaming SQL Sink Docs
+
+This directory documents sink connectors for Streaming SQL.
+
+## Support matrix
+
+| Connector | Supported formats |
+|---|---|
+| `kafka` | `json` / `raw_string` / `raw_bytes` |
+| `filesystem` | `csv` / `parquet` / `json`(JSONL) / `avro` / `orc` |
+| `s3` | `csv` / `parquet` |
+| `delta` | `csv` / `parquet` / `json`(JSONL) / `avro` / `orc` |
+| `iceberg` | `csv` / `parquet` |
+| `lanceDB` | `lance` |
+
+## Documents
+
+- [Kafka Sink](kafka-sink.md)
+- [Filesystem Sink](filesystem-sink.md)
+- [S3 Sink](s3-sink.md)
+- [Delta Sink](delta-sink.md)
+- [Iceberg Sink](iceberg-sink.md)
+- [LanceDB Sink](lancedb-sink.md)
+
+## Notes
+
+- Configure sink connectors via `WITH (...)` in `CREATE STREAMING TABLE ... AS SELECT ...`.
+- Use `type='sink'` explicitly for sink tables.
+- Only `lanceDB` accepts `format='lance'`.
+- For file-like sinks, `format='json'` is written as JSON Lines (NDJSON, `.jsonl`).
+

docs/streaming-sql/Sink/README.md

@@ -0,0 +1,33 @@
+# Delta Sink
+
+`delta` connector 对应 Delta 数据湖写出通道。
+
+## 支持格式
+
+- `csv`
+- `parquet`
+- `json`（写出为 JSON Lines / NDJSON，文件后缀 `.jsonl`）
+- `avro`
+- `orc`
+
+## 常用 WITH 参数
+
+- `connector='delta'`
+- `type='sink'`
+- `format='csv'|'parquet'|'json'|'avro'|'orc'`
+- `path='/data/delta/orders'`（本地）或对象存储前缀
+- 可选 S3 参数：`s3.bucket` / `s3.region` / `s3.endpoint` / AKSK
+- `parquet.compression`（仅 `parquet`）
+
+## 示例（CREATE STREAMING TABLE）
+
+```sql
+CREATE STREAMING TABLE st_delta_parquet
+WITH (
+  connector='delta',
+  type='sink',
+  format='parquet',
+  path='/tmp/delta_orders'
+) AS
+SELECT * FROM src_kafka_orders;
+```