You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tsfile.apache.org by "JackieTien97 (via GitHub)" <gi...@apache.org> on 2026/06/15 03:32:20 UTC
[I] [Feature] Add Spark TsFile connector support for the TsFile table model [tsfile]
JackieTien97 opened a new issue, #843:
URL: https://github.com/apache/tsfile/issues/843
## Motivation
Apache TsFile already has ecosystem integration with Spark, and the tree-model Spark TsFile connector can be used as a reference:
https://github.com/apache/iotdb-extras/tree/master/connectors/spark-tsfile
TsFile now also supports the table model, including `TableSchema`, `ColumnCategory.TAG/FIELD`, table-model read APIs, and table-model write APIs such as `TsFileWriter#registerTableSchema` and `TsFileWriter#writeTable`.
We need a Spark connector for table-model TsFile so users can read and write table-model TsFiles directly through Spark SQL/DataFrame APIs.
## Goal
Develop a Spark SQL/DataFrame connector for TsFile table model. The connector should reuse existing TsFile Java read/write APIs as much as possible, instead of duplicating TsFile parsing or writing logic.
## Expected Scope
The initial implementation should support:
- Reading table-model TsFile files or directories into Spark DataFrames.
- Inferring or loading table schemas from TsFile metadata, including:
- table name
- time column
- TAG columns
- FIELD columns
- TsFile data types and corresponding Spark SQL types
- Preserving table-model semantics:
- TAG columns identify devices
- FIELD columns represent measurements
- null values and sparse field values are handled correctly
- Reading multiple TsFiles with compatible schemas.
- Column pruning where possible.
- Predicate pushdown where possible, especially:
- time-range filters
- tag filters
- Writing Spark DataFrames into table-model TsFiles, with options such as:
- table name
- tag columns
- field columns
- encoding/compression defaults if needed
- Providing user-facing examples for Spark SQL/DataFrame read and write workflows.
## Proposed User Experience
Example read API:
```scala
val df = spark.read
.format("tsfile")
.option("model", "table")
.option("table", "weather")
.load("/path/to/tsfile-dir")
df.select("time", "city", "device", "temperature")
.where("city = 'beijing'")
.show()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tsfile.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Add Spark TsFile connector support for the TsFile table model [tsfile]
Posted by "DaZuiZui (via GitHub)" <gi...@apache.org>.
DaZuiZui commented on issue #843:
URL: https://github.com/apache/tsfile/issues/843#issuecomment-4714317540
## 1. Goal
Add a Spark connector for the TsFile table model. The proposed format name is `tsfile`, so users can read and write table-model TsFile data through Spark DataFrame and Spark SQL APIs.
For the first implementation, the connector will explicitly target:
- **Spark 3.x DataSource V2 only**.
- Spark V2 APIs such as `TableProvider`, `ScanBuilder`, and `BatchWrite`.
- No Spark 2.4 DataSource V1 compatibility.
- The old `iotdb-extras spark-tsfile` module may be used as a reference for user experience and tests, but its code should not be copied directly.
The connector should reuse existing TsFile Java APIs as much as possible:
- Read path: `DeviceTableModelReader`, `TsFileSequenceReader#getTableSchemaMap`, `TableQueryExecutor`, etc.
- Write path: `TsFileWriter#registerTableSchema`, `TsFileWriter#writeTable`.
- The Spark connector should not reimplement TsFile binary parsing, metadata parsing, page/chunk decoding, or writing logic.
## 2. Feature Definition
One Spark row maps to one TsFile table-model row:
```text
row := (time, tag_1, tag_2, ..., tag_n, field_1, ..., field_m)
```
| Component | Meaning |
| ------------------ | ------------------------------------------------------------ |
| `table` | Logical table name in the TsFile table model. It corresponds to `TableSchema#getTableName()` and is specified through an option. By default, it is not exposed as a DataFrame column. |
| `time` | Time column. It corresponds to the TsFile timestamp and is not a normal `TableSchema` measurement column. The default DataFrame column name is `time`. |
| `TAG` column | Device identity column. It corresponds to `ColumnCategory.TAG`. Multiple TAG columns form the device identity in order. The first implementation only supports non-null `StringType` TAG columns. |
| `FIELD` column | Measurement/value column. It corresponds to `ColumnCategory.FIELD`. Sparse values and nulls are allowed. |
| `ATTRIBUTE` column | `ColumnCategory.ATTRIBUTE`. Not supported in the first implementation; the connector should fail with a clear error if encountered. |
| `TIME` category | `ColumnCategory.TIME`. Not used as the DataFrame time column in the first implementation; the connector should fail with a clear error if encountered. |
In the current TsFile table model, table and column names are normalized to lower case. Therefore, schema lookup and compatibility checks inside the connector should use lower-case names.
## 3. Initial Scope
The first implementation supports:
- Reading a single TsFile file.
- Reading a TsFile directory or glob path.
- Reading multiple TsFiles with compatible schemas.
- Inferring Spark schema from TsFile metadata.
- Selecting a specific table.
- Column pruning.
- Predicate pushdown:
- `time` range filters.
- `time = c`.
- TAG equality filters, for example `city = 'beijing'`.
- `AND` combinations of the supported predicates above.
- Returning null values correctly for sparse FIELD columns.
- Writing table-model TsFile data from a Spark DataFrame.
- Configuring table name, time column, TAG columns, FIELD columns, encoding, and compression.
- Providing Scala DataFrame and Spark SQL examples.
The first implementation does not support:
- Tree-model Spark connector behavior.
- Spark 2.4 DataSource V1.
- Spark structured streaming.
- Update/delete/merge-on-write.
- Reading multiple tables into one DataFrame.
- FIELD value predicate pushdown.
- Complex `OR` predicate pushdown.
- Non-string TAG values or null TAG values.
- `ATTRIBUTE`, `TIME`, `VECTOR`, `UNKNOWN`, or arbitrary `OBJECT` values.
## 4. User Syntax
### DataFrame Read
```scala
val df = spark.read
.format("tsfile")
.option("model", "table")
.option("table", "weather")
.load("/data/weather-tsfile")
df.select("time", "city", "device", "temperature")
.where("time >= 1704067200000 and city = 'beijing'")
.show()
```
### Spark SQL Read
This is Spark SQL over the Spark connector. TsFile does not need to implement its own SQL parser or SQL engine.
```sql
CREATE TEMPORARY VIEW weather
USING tsfile
OPTIONS (
path '/data/weather-tsfile',
model 'table',
table 'weather'
);
SELECT time, city, device, temperature
FROM weather
WHERE city = 'beijing';
```
### DataFrame Write
```scala
df.write
.format("tsfile")
.option("model", "table")
.option("table", "weather")
.option("timeColumn", "time")
.option("tagColumns", "city,device")
.option("fieldColumns", "temperature,humidity,status")
.option("compression", "LZ4")
.mode("append")
.save("/data/weather-tsfile")
```
## 5. Main Options
| Option | Meaning |
| -------------------- | ------------------------------------------------------------ |
| `model` | Fixed to `table`. |
| `table` | Table name. Required for writing; required when reading files containing multiple tables. |
| `timeColumn` | DataFrame column used as the timestamp column. Default is `time`. |
| `tagColumns` | Required for writing. Comma-separated list. The order defines the device identity. |
| `fieldColumns` | Optional for writing. By default, all non-time and non-TAG columns are FIELD columns. |
| `timestampAs` | Read the time column as `long` or `timestamp`. Default is `long`. |
| `timestampPrecision` | `ms`, `us`, or `ns`. Used when converting between Spark `TimestampType` and raw TsFile timestamps. |
| `mergeSchema` | Whether to merge FIELD schemas when reading multiple files. Default is false. |
| `pushdown` | Whether to enable predicate pushdown. Default is true. |
| `compression` | Compression type for writing, for example `LZ4`, `SNAPPY`, `GZIP`, or `ZSTD`. |
| `encoding` | Encoding type for writing. It must be validated against TsFile encoding/type compatibility rules. |
| `nullTagPolicy` | The first implementation only supports `error`, meaning null TAG values fail immediately. |
## 6. Data Type Support
| TsFile type | Spark type | Support |
| ----------- | ----------------------------- | ------------------------------------------- |
| `BOOLEAN` | `BooleanType` | FIELD supported |
| `INT32` | `IntegerType` | FIELD supported |
| `INT64` | `LongType` | FIELD supported |
| `FLOAT` | `FloatType` | FIELD supported |
| `DOUBLE` | `DoubleType` | FIELD supported |
| `TEXT` | `StringType` | FIELD supported |
| `STRING` | `StringType` | FIELD/TAG supported; TAG must use this type |
| `DATE` | `DateType` or `IntegerType` | FIELD supported |
| `TIMESTAMP` | `TimestampType` or `LongType` | FIELD supported |
| `BLOB` | `BinaryType` | FIELD supported |
| `VECTOR` | none | Not supported |
| `UNKNOWN` | none | Not supported |
| `OBJECT` | TBD | Not supported in the first implementation |
TAG columns are strictly limited to non-null `StringType` in the first implementation. The reason is that the current table model builds device identity from the table name and TAG values. `Tablet#getDeviceID` generates device identifiers from TAG values, and the TAG filter API also uses string-based semantics. To avoid unstable identity and inconsistent filter semantics, the first implementation rejects non-string TAG values and null TAG values.
## 7. Read Semantics
When reading, the connector should:
- Support a single file, a directory, or a glob path.
- Use `TsFileSequenceReader#getTableSchemaMap` to read table metadata.
- If a file contains only one table, infer the table automatically.
- If a file contains multiple tables, require the user to specify `option("table", "...")`.
- By default, require strict schema compatibility across multiple files: table name, column name, `ColumnCategory`, and `TSDataType` must match.
- When `mergeSchema=true`, only allow FIELD column union. TAG columns must still have the same order and type.
- Fail clearly when encountering `ATTRIBUTE`, `TIME`, non-string TAG columns, or unsupported types.
- Return Spark SQL null for missing FIELD values.
### Column Pruning Risk
The current table reader may depend on at least one FIELD/measurement column when generating rows. Therefore, a query such as:
```scala
df.select("time", "city")
```
may require special handling:
- Either improve the reader to support time/TAG-only projection.
- Or let the connector internally read one minimal FIELD column and drop it before returning Spark rows.
This should be covered by an integration test.
## 8. Predicate Pushdown
The first implementation guarantees the following pushdown support:
| Spark predicate | TsFile pushdown |
| ------------------------------------------ | ------------------- |
| `time = c` | time filter |
| `time > c`, `time >= c` | time filter |
| `time < c`, `time <= c` | time filter |
| `time BETWEEN a AND b` | time filter |
| `tag = 'x'` | TAG equality filter |
| `AND` combinations of supported predicates | combined pushdown |
The following predicates remain Spark residual filters:
- `OR`.
- Complex functions.
- FIELD value predicates.
- TAG `!=`, `IN`, range comparison, and null check.
- Non-string TAG comparisons.
## 9. Write Semantics
Spark writes must follow these rules:
- Each Spark task/partition writes an independent `part-*.tsfile` file.
- Output files are committed through Spark's commit protocol.
- Multiple tasks must not append concurrently to the same physical TsFile.
- `append` means appending new TsFile files to the target directory, not appending data into existing TsFile files.
Write flow:
1. Read `timeColumn`.
2. If it is Spark `TimestampType`, convert it to raw TsFile long using `timestampPrecision`. If it is `LongType`, use it directly.
3. Read TAG columns in the configured order.
4. Build a `Tablet` containing TAG and FIELD columns.
5. Write null FIELD values as null/sparse values.
6. Call `TsFileWriter#writeTable` when `maxRowsPerTablet` is reached or when the task finishes.
7. Call `TsFileWriter#registerTableSchema` before writing.
8. Reject non-string TAG values or null TAG values immediately.
## 10. Module and Components
A new module is proposed:
```text
java/spark-tsfile
```
It should be registered under the Java aggregator in `java/pom.xml`.
Dependency principles:
- Spark/Hadoop dependencies should only live in the `java/spark-tsfile` submodule.
- Mark Spark/Hadoop dependencies as `provided` where possible.
- Avoid polluting the core TsFile library.
- Avoid dependency convergence or duplicate-class issues in the root POM.
Main components:
| Component | Responsibility |
| ----------------------------- | ------------------------------------------------------------ |
| `TsFileTableProvider` | Spark DataSource V2 entry point. Implements `TableProvider` and registers the short name `tsfile`. |
| `TsFileTableOptions` | Parses and validates connector options. |
| `TsFileTableSchemaInferer` | Infers Spark `StructType` from TsFile metadata. |
| `TsFileTableTypeConverter` | Converts between TsFile `TSDataType` and Spark SQL types. |
| `TsFileTableScanBuilder` | Spark V2 `ScanBuilder`; handles column pruning and pushdown. |
| `TsFileTablePartitionReader` | Reads TsFile data and converts it to Spark rows. |
| `TsFileTableFilterTranslator` | Converts Spark filters to TsFile time/TAG filters and records residual filters. |
| `TsFileTableWriteBuilder` | Spark V2 write builder; validates write schema. |
| `TsFileTableBatchWrite` | Spark V2 `BatchWrite`; coordinates task writers and commit. |
| `TsFileTablePartitionWriter` | Converts Spark rows to `Tablet` and writes independent `part-*.tsfile` files. |
## 11. Implementation Phases
1. **Read path**
- Add the `java/spark-tsfile` module.
- Add Spark 3.x V2 dependencies. Keep Spark/Hadoop dependencies `provided` where possible.
- Implement `TableProvider`.
- Implement option parsing.
- Implement single-file, single-table schema inference.
- Implement basic DataFrame read.
- Reject unsupported schemas, such as non-string TAG columns, `ATTRIBUTE`, and `TIME`.
2. **Directory and multi-file read**
- Support directories and glob paths.
- Validate schema compatibility across multiple files.
- Support strict `mergeSchema=false` mode.
- Add tests for compatible and incompatible schemas.
3. **Projection and pushdown**
- Implement column pruning.
- Handle the time/TAG-only projection risk.
- Implement time range, `time = c`, and TAG equality pushdown.
- Implement `AND` combinations of supported predicates.
- Keep unsupported filters as residual filters.
4. **Write path**
- Implement DataFrame write.
- Build and register `TableSchema`.
- Implement Spark V2 `BatchWrite`.
- Make each task write independent `part-*.tsfile` files.
- Support FIELD null values.
- Reject non-string TAG values and null TAG values.
- Add Spark DataFrame -> TsFile -> Spark DataFrame round-trip tests.
5. **Examples and documentation**
- Add Scala DataFrame read/write examples.
- Add Spark SQL read/CTAS examples.
- Document options and type mapping.
- Add a small table-model TsFile test fixture.
## 12. Test Plan
The tests should cover:
- Option parsing:
- Missing `table` fails.
- Missing `tagColumns` fails.
- Duplicate column names after lower-case normalization fail.
- Invalid `timestampPrecision` fails.
- Non-string TAG columns fail.
- Type mapping:
- Supported types map correctly.
- Unsupported types fail with clear errors.
- `ATTRIBUTE` / `TIME` categories fail with clear errors.
- Filter translation:
- Time comparisons.
- TAG equality.
- `AND` combinations of supported predicates.
- Unsupported predicates remain residual filters.
- Integration tests:
- Read a single file with a single table.
- Read a multi-table file with a specified table.
- Read multiple files from a directory.
- Sparse FIELD values return null.
- Selected columns only.
- Time/TAG-only projection.
- Pushdown correctness.
- Each Spark task writes an independent `part-*.tsfile`.
- Round-trip write/read correctness.
## 13. Initial Implementation Decisions
The first implementation uses conservative defaults:
- Timestamp values are exposed as raw `LongType` by default. Spark `TimestampType` is only used when users explicitly set `timestampAs=timestamp`.
- `mergeSchema=true` only unions compatible FIELD columns. It does not perform numeric widening; same-name columns with different TsFile types fail fast.
- The first implementation only supports global `encoding` and `compression` options. Per-column encoding/compression can be added later if needed.
Feedback on Spark version compatibility, connector API semantics, and implementation scope is welcome.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Add Spark TsFile connector support for the TsFile table model [tsfile]
Posted by "wipheg (via GitHub)" <gi...@apache.org>.
wipheg commented on issue #843:
URL: https://github.com/apache/tsfile/issues/843#issuecomment-4708049904
hi @JackieTien97 , i'd like to work on this issue. could you assign this to me?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Add Spark TsFile connector support for the TsFile table model [tsfile]
Posted by "DaZuiZui (via GitHub)" <gi...@apache.org>.
DaZuiZui commented on issue #843:
URL: https://github.com/apache/tsfile/issues/843#issuecomment-4764391552
Hi @JackieTien97 ,
I have finished the initial implementation of the Spark TsFile table model connector for #843.
The PR is ready for review:
https://github.com/apache/tsfile/pull/845
This PR adds a Spark DataSource V2 connector for reading and writing table-model TsFile files with .format("tsfile"). It includes schema inference, append writes, column pruning, basic predicate pushdown, mergeSchema=true, CTAS support, examples, and integration tests.
I have run the relevant Maven tests locally:
**JAVA_HOME=$(/usr/libexec/java_home -v 21) ./mvnw -P with-java -pl java/spark-tsfile -am test**
The build passed successfully.
Please take a look when you have time. Any feedback is welcome.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Add Spark TsFile connector support for the TsFile table model [tsfile]
Posted by "JackieTien97 (via GitHub)" <gi...@apache.org>.
JackieTien97 commented on issue #843:
URL: https://github.com/apache/tsfile/issues/843#issuecomment-4715526477
> hi [@JackieTien97](https://github.com/JackieTien97) , i'd like to work on this issue. could you assign this to me?
Hi @wipheg , since @DaZuiZui has already put a more detailed implementation plan, I would prefer to assign this issue to hime
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature] Add Spark TsFile connector support for the TsFile table model [tsfile]
Posted by "DaZuiZui (via GitHub)" <gi...@apache.org>.
DaZuiZui commented on issue #843:
URL: https://github.com/apache/tsfile/issues/843#issuecomment-4715694370
@wipheg Thanks for offering to help! I'm currently working on this issue, but I'll let you know if I need any input.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org