This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 83f76f2d52d feat: add Apache TsFile input format extension (#20729)
83f76f2d52d is described below
commit 83f76f2d52da8f0c53d13fef0fb3e6ef8436f57f
Author: Colin Lee <[email protected]>
AuthorDate: Sun Nov 1 10:46:33 2026 +0800
feat: add Apache TsFile input format extension (#20729)
* feat: add Apache TsFile input format extension
Adds extensions-contrib/tsfile-extensions, a community extension that lets
batch ingestion read Apache TsFile files through the TsFile Java table
model.
- InputFormat type "tsfile" with optional tables, columns, and interval.
- interval is pushed down to the TsFile reader, which skips chunks outside
the range using the time statistics stored with each chunk.
- Ships a sample TsFile, an ingestion spec, and documentation under
docs/development/extensions-contrib/tsfile.md.
* build: bump tsfile extension parent to 40.0.0-SNAPSHOT
Druid master moved to 40.0.0-SNAPSHOT, so the module's parent version has to
match; otherwise Maven cannot resolve the parent POM.
* docs: add spelling suppressions for the TsFile input format docs
* build: fix strict compilation and dependency analysis for tsfile extension
- Drop the unused throws clause that Error Prone rejects under
strictCompile.
- Manage the xz and lz4-java overrides instead of declaring them, and remove
dependencies the module does not use, so maven-dependency-plugin:analyze
passes.
* fix: address review feedback on the tsfile input format
- Reject files with no table schema instead of silently reading zero rows,
so
tree-model files fail with a clear error.
- Apply the timestamp alias after the measurement columns so a TsFile column
named like the timestampSpec column cannot override the file's own time.
- Close the TsFile reader when table validation fails.
* build: bump xz to 1.10 and drop the tsfile module's version overrides
TsFile 2.5.0 requires xz 1.10, so upgrade it in the root pom instead of
overriding it in the extension. lz4-java is already at 1.12.0 upstream, so
the
module needs no dependencyManagement at all.
* build: update the xz version in licenses.yaml to 1.10
The root pom now manages xz 1.10, which packaging-check compares against the
license manifest.
* build: quote the xz version in licenses.yaml
YAML parses an unquoted 1.10 as the float 1.1, so the manifest did not match
the resolved artifact version.
* test: add end-to-end ingestion tests for the tsfile input format
Covers batch ingestion with a local input source and MSQ INSERT ... SELECT
from TABLE(EXTERN(...)) against a TsFile written by the test.
* build: manage the tsfile version in the root pom
The embedded tests use TsFile writer classes directly, so declare the
dependency there too. Manage the version once in the root pom instead of
repeating it in each module.
* docs: add a SQL-based ingestion example for the tsfile input format
* fix: register the tsfile dependencies and honor the column projection
- Add an Apache TsFile entry to licenses.yaml so the contrib bundle carries
the licenses and notice for org.apache.tsfile:tsfile and :common.
- When columns is omitted, project the physical columns using the ingestion
schema's ColumnsFilter, and for EXTERN (ColumnsFilter.all()) use the fixed
dimension list that carries the external table's column signature.
- Fall back to reading every column when the projection would only contain
tag
columns, since the TsFile table reader returns no rows in that case.
* fix: keep the ingestion column filter authoritative for projections
- Use the fixed dimension list (the EXTERN signature) only when the
ColumnsFilter carries no information. A batch filter that happens to cover
every physical column now stays authoritative, so transform inputs and
aggregator inputs are no longer dropped.
- Remove the metric-output-names fallback, which cannot stand in for the
aggregator input columns.
- Stop widening tag-only projections to every physical column. A single
supported measurement column is read as a row carrier and kept out of the
row, so an omitted unsupported column is never read.
---
distribution/pom.xml | 2 +
docs/configuration/extensions.md | 1 +
docs/development/extensions-contrib/tsfile.md | 167 ++++++++
embedded-tests/pom.xml | 11 +
.../embedded/tsfile/TsFileIngestionTest.java | 205 +++++++++
extensions-contrib/tsfile-extensions/README.md | 123 ++++++
.../tsfile-extensions/examples/data/sensor.tsfile | Bin 0 -> 1463 bytes
.../tsfile-extensions/examples/sensor-ingest.json | 49 +++
extensions-contrib/tsfile-extensions/pom.xml | 98 +++++
.../data/input/tsfile/TsFileExtensionsModule.java | 48 +++
.../data/input/tsfile/TsFileInputEntityReader.java | 462 +++++++++++++++++++++
.../druid/data/input/tsfile/TsFileInputFormat.java | 148 +++++++
.../org.apache.druid.initialization.DruidModule | 16 +
.../data/input/tsfile/TsFileInputFormatTest.java | 367 ++++++++++++++++
.../druid/data/input/tsfile/TsFileTestData.java | 149 +++++++
licenses.yaml | 17 +
pom.xml | 8 +
website/.spelling | 10 +
18 files changed, 1881 insertions(+)
diff --git a/distribution/pom.xml b/distribution/pom.xml
index 7202d8910f8..ff0ead00c6a 100644
--- a/distribution/pom.xml
+++ b/distribution/pom.xml
@@ -461,6 +461,8 @@
<argument>-c</argument>
<argument>org.apache.druid.extensions.contrib:druid-thrift-extensions</argument>
<argument>-c</argument>
+
<argument>org.apache.druid.extensions.contrib:druid-tsfile-extensions</argument>
+ <argument>-c</argument>
<argument>org.apache.druid.extensions.contrib:druid-time-min-max</argument>
<argument>-c</argument>
<argument>org.apache.druid.extensions.contrib:druid-virtual-columns</argument>
diff --git a/docs/configuration/extensions.md b/docs/configuration/extensions.md
index 89102efa4ec..e4b485f25c5 100644
--- a/docs/configuration/extensions.md
+++ b/docs/configuration/extensions.md
@@ -95,6 +95,7 @@ All of these community extensions can be downloaded using
[pull-deps](../operati
|kafka-emitter|Kafka metrics
emitter|[link](../development/extensions-contrib/kafka-emitter.md)|
|openlineage-emitter|OpenLineage data lineage
emitter|[link](../development/extensions-contrib/openlineage-emitter.md)|
|druid-thrift-extensions|Support thrift ingestion
|[link](../development/extensions-contrib/thrift.md)|
+|druid-tsfile-extensions|Support for ingesting Apache TsFile
files|[link](../development/extensions-contrib/tsfile.md)|
|druid-opentsdb-emitter|OpenTSDB metrics emitter
|[link](../development/extensions-contrib/opentsdb-emitter.md)|
|druid-moving-average-query|Support for [Moving
Average](https://en.wikipedia.org/wiki/Moving_average) and other Aggregate
[Window
Functions](https://en.wikibooks.org/wiki/Structured_Query_Language/Window_functions)
in Druid
queries.|[link](../development/extensions-contrib/moving-average-query.md)|
|druid-influxdb-emitter|InfluxDB metrics
emitter|[link](../development/extensions-contrib/influxdb-emitter.md)|
diff --git a/docs/development/extensions-contrib/tsfile.md
b/docs/development/extensions-contrib/tsfile.md
new file mode 100644
index 00000000000..21a9240e048
--- /dev/null
+++ b/docs/development/extensions-contrib/tsfile.md
@@ -0,0 +1,167 @@
+---
+id: tsfile
+title: "Apache TsFile input format"
+---
+
+<!--
+ ~ 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.
+ -->
+
+To use this Apache Druid extension,
[include](../../configuration/extensions.md#loading-extensions)
`druid-tsfile-extensions` in the extensions load list.
+
+This extension enables Druid to ingest [Apache
TsFile](https://tsfile.apache.org/) files. TsFile is the columnar
+storage format for time series data used by [Apache
IoTDB](https://iotdb.apache.org/). It is read through the
+TsFile Java library, version 2.5.0.
+
+The extension supports batch ingestion from local files and from any input
source that Druid can read as a file
+(for example, S3). Local files are read in place; files from remote input
sources are first fetched to the task's
+temporary directory, so a non-local input source needs enough local disk for
one file at a time.
+
+## TsFile input format
+
+| Field | Type | Description | Required |
+|-------|------|-------------|----------|
+| type | String | Must be `tsfile` | yes |
+| tables | Array of String | Tables to read. When omitted or empty, all tables
in the file are read. | no |
+| columns | Array of String | Columns to read. When omitted or empty, the
columns implied by the ingestion schema are read: the dimensions and metrics of
the ingestion spec, or the external table's column signature for SQL ingestion.
All columns are read when the schema does not pin them down. | no |
+| interval | String | Time range to read, in Druid interval format
(`start/end`). The range is end-exclusive. When omitted, each file is read in
full. | no |
+
+### Time range pruning
+
+When `interval` is set, it is pushed down to the TsFile reader, which uses the
time statistics stored with each
+chunk to skip chunks that fall outside the range. Only matching rows are
returned.
+
+Druid does not pass the task's time window to input formats, so the range has
to be set explicitly here. Rows are
+still filtered by the granularity spec as usual.
+
+### Timestamp
+
+TsFile stores its own timestamp, which is exposed to Druid under the column
name `Time`. The ingestion spec must
+therefore select `Time` as the timestamp column with a numeric format, for
example:
+
+```json
+"timestampSpec": {
+ "column": "Time",
+ "format": "millis"
+}
+```
+
+If `timestampSpec.column` is set to another name, the TsFile timestamp is also
placed under that name so that
+existing ingestion specs keep working. In that case a TsFile column with the
same name is shadowed.
+
+### Column types
+
+Tag, field, and attribute columns are all converted to input rows. The mapping
of TsFile data types is:
+
+| TsFile type | Druid value |
+|-------------|-------------|
+| INT32 | Integer |
+| INT64 | Long |
+| FLOAT | Float |
+| DOUBLE | Double |
+| BOOLEAN | Boolean |
+| TEXT, STRING | String |
+| TIMESTAMP | Long (epoch milliseconds) |
+| DATE | Integer (epoch day) |
+| BLOB | Base64-encoded String |
+
+`VECTOR`, `OBJECT`, `ROW`, and `UNKNOWN` columns are not supported; requesting
one of them fails the ingestion task.
+Null values are omitted from the row.
+
+## Example: local batch ingestion
+
+Suppose a TsFile file `sensor.tsfile` contains a table `sensor` with the tag
columns `region` and `device` and the
+value columns `temperature`, `humidity`, and `status`. The following spec
ingests it:
+
+```json
+{
+ "type": "index_parallel",
+ "spec": {
+ "ioConfig": {
+ "type": "index_parallel",
+ "inputSource": {
+ "type": "local",
+ "baseDir": "/data/tsfile",
+ "filter": "*.tsfile"
+ },
+ "inputFormat": {
+ "type": "tsfile",
+ "tables": ["sensor"],
+ "columns": ["region", "device", "temperature", "humidity", "status"]
+ }
+ },
+ "dataSchema": {
+ "dataSource": "sensor",
+ "timestampSpec": {
+ "column": "Time",
+ "format": "millis"
+ },
+ "dimensionsSpec": {
+ "dimensions": ["region", "device"]
+ },
+ "metricsSpec": [
+ {"type": "doubleSum", "name": "temperature", "fieldName":
"temperature"},
+ {"type": "doubleSum", "name": "humidity", "fieldName": "humidity"}
+ ],
+ "granularitySpec": {
+ "type": "uniform",
+ "segmentGranularity": "DAY",
+ "queryGranularity": "MINUTE",
+ "rollup": true
+ }
+ },
+ "tuningConfig": {
+ "type": "index_parallel"
+ }
+ }
+}
+```
+
+## Example: SQL-based ingestion
+
+The same file can be loaded with an MSQ `INSERT ... SELECT` statement.
`EXTERN` takes the input source spec, the
+input format spec, and the external table's column signature:
+
+```sql
+INSERT INTO sensor
+SELECT
+ MILLIS_TO_TIMESTAMP("Time") AS __time,
+ region,
+ device,
+ temperature
+FROM TABLE(
+ EXTERN(
+ '{"type":"local","baseDir":"/data/tsfile","filter":"*.tsfile"}',
+ '{"type":"tsfile","tables":["sensor"]}',
+ '[{"name":"Time","type":"long"},{"name":"region","type":"string"},'
+ '{"name":"device","type":"string"},{"name":"temperature","type":"double"}]'
+ )
+)
+PARTITIONED BY DAY
+```
+
+TsFile's implicit `Time` column is declared as `long` in the signature and
converted to `__time` with
+`MILLIS_TO_TIMESTAMP`. Columns that are not listed in the signature are not
read.
+
+## Limitations
+
+- The input format is not splittable. With a splittable input source such as
`local`, `s3`, or `gcs`, Druid still
+ parallelizes ingestion one file per task.
+- Druid's segment granularity is not pushed down to TsFile. To read only part
of a file, set `interval`.
+- A single ingestion spec reads one row schema. If a file contains several
tables with different columns, either
+ select one table with `tables` or ingest each table separately.
diff --git a/embedded-tests/pom.xml b/embedded-tests/pom.xml
index 33e97ece929..9b1766c4765 100644
--- a/embedded-tests/pom.xml
+++ b/embedded-tests/pom.xml
@@ -376,6 +376,17 @@
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.druid.extensions.contrib</groupId>
+ <artifactId>druid-tsfile-extensions</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tsfile</groupId>
+ <artifactId>tsfile</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.druid.extensions</groupId>
<artifactId>druid-hdfs-storage</artifactId>
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/tsfile/TsFileIngestionTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/tsfile/TsFileIngestionTest.java
new file mode 100644
index 00000000000..f5c1982c352
--- /dev/null
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/tsfile/TsFileIngestionTest.java
@@ -0,0 +1,205 @@
+/*
+ * 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.druid.testing.embedded.tsfile;
+
+import org.apache.druid.data.input.tsfile.TsFileExtensionsModule;
+import org.apache.druid.data.input.tsfile.TsFileInputFormat;
+import org.apache.druid.indexing.common.task.IndexTask;
+import org.apache.druid.indexing.common.task.TaskBuilder;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.aggregation.DoubleSumAggregatorFactory;
+import org.apache.druid.query.http.SqlTaskStatus;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedClusterApis;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedHistorical;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.apache.druid.testing.embedded.msq.EmbeddedMSQApis;
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.file.metadata.ColumnSchemaBuilder;
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.v4.ITsFileWriter;
+import org.apache.tsfile.write.v4.TsFileWriterBuilder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * End-to-end ingestion tests for {@link TsFileInputFormat} running inside an
embedded Druid cluster.
+ * The TsFile written by {@link #createTsFile()} holds three rows of one
table, mirroring the sample file
+ * shipped with the extension.
+ */
+public class TsFileIngestionTest extends EmbeddedClusterTestBase
+{
+ private static final String TABLE_NAME = "sensor";
+ private static final int ROW_COUNT = 3;
+
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator();
+ private final EmbeddedIndexer indexer = new EmbeddedIndexer()
+ .setServerMemory(300_000_000L)
+ .addProperty("druid.worker.capacity", "2");
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+
+ @TempDir
+ private File tempDir;
+
+ @Override
+ protected EmbeddedDruidCluster createCluster()
+ {
+ return EmbeddedDruidCluster
+ .withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .addExtension(TsFileExtensionsModule.class)
+ .addServer(overlord)
+ .addServer(coordinator)
+ .addServer(indexer)
+ .addServer(broker)
+ .addServer(new EmbeddedHistorical());
+ }
+
+ /**
+ * Batch ingestion driven by a task spec that reads a local TsFile.
+ */
+ @Test
+ @Timeout(180)
+ public void testBatchIngestionTask() throws IOException,
WriteProcessException
+ {
+ final IndexTask task = TaskBuilder
+ .ofTypeIndex()
+ .localInputSourceWithFiles(createTsFile())
+ .inputFormat(new TsFileInputFormat(null, null, null))
+ .timestampColumn(TsFileInputFormat.TIME_COLUMN_NAME)
+ .dimensions("region", "device")
+ .metricAggregates(new DoubleSumAggregatorFactory("temperature",
"temperature"))
+ .segmentGranularity("DAY")
+ .dataSource(dataSource)
+ .withId(EmbeddedClusterApis.newTaskId(dataSource));
+
+ cluster.callApi().runTask(task, overlord);
+ cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
+
+ assertIngestedRows();
+ }
+
+ /**
+ * MSQ {@code INSERT ... SELECT FROM TABLE(EXTERN(...))} reading the same
TsFile.
+ */
+ @Test
+ @Timeout(180)
+ public void testInsertSelectFromExtern() throws IOException,
WriteProcessException
+ {
+ final File tsFile = createTsFile();
+ final EmbeddedMSQApis msqApis = new EmbeddedMSQApis(cluster, overlord);
+
+ // Escape backslashes so a Windows path stays valid inside the JSON input
source spec.
+ final String baseDir =
StringUtils.replace(tsFile.getParentFile().getAbsolutePath(), "\\", "\\\\");
+
+ final String sql = StringUtils.format(
+ "INSERT INTO %s\n"
+ + "SELECT\n"
+ + " MILLIS_TO_TIMESTAMP(\"Time\") AS __time,\n"
+ + " region,\n"
+ + " device,\n"
+ + " temperature\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + "
'{\"type\":\"local\",\"baseDir\":\"%s\",\"filter\":\"*.tsfile\"}',\n"
+ + " '{\"type\":\"tsfile\",\"tables\":[\"%s\"]}',\n"
+ + "
'[{\"name\":\"Time\",\"type\":\"long\"},{\"name\":\"region\",\"type\":\"string\"},"
+ +
"{\"name\":\"device\",\"type\":\"string\"},{\"name\":\"temperature\",\"type\":\"double\"}]'\n"
+ + " )\n"
+ + ")\n"
+ + "PARTITIONED BY DAY",
+ dataSource,
+ baseDir,
+ TABLE_NAME
+ );
+
+ final SqlTaskStatus taskStatus = msqApis.submitTaskSql(sql);
+ cluster.callApi().waitForTaskToSucceed(taskStatus.getTaskId(), overlord);
+ cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
+
+ assertIngestedRows();
+ cluster.callApi().verifySqlQuery(
+ "SELECT SUM(temperature) FROM %s",
+ dataSource,
+ String.valueOf(21.5 + 22.5 + 23.5)
+ );
+ }
+
+ private void assertIngestedRows()
+ {
+ cluster.callApi().verifySqlQuery(
+ "SELECT COUNT(*) FROM %s",
+ dataSource,
+ String.valueOf(ROW_COUNT)
+ );
+ cluster.callApi().verifySqlQuery(
+ "SELECT region, COUNT(*) FROM %s GROUP BY region ORDER BY region",
+ dataSource,
+ "beijing,2\nshanghai,1"
+ );
+ cluster.callApi().verifySqlQuery(
+ "SELECT MIN(__time), MAX(__time) FROM %s",
+ dataSource,
+ "1970-01-01T00:00:01.000Z,1970-01-01T00:00:01.002Z"
+ );
+ }
+
+ private File createTsFile() throws IOException, WriteProcessException
+ {
+ final File file = new File(tempDir, "sensor.tsfile");
+ final TableSchema tableSchema = new TableSchema(
+ TABLE_NAME,
+ Arrays.asList(
+ new
ColumnSchemaBuilder().name("region").dataType(TSDataType.STRING).category(ColumnCategory.TAG).build(),
+ new
ColumnSchemaBuilder().name("device").dataType(TSDataType.STRING).category(ColumnCategory.TAG).build(),
+ new
ColumnSchemaBuilder().name("temperature").dataType(TSDataType.FLOAT).build()
+ )
+ );
+
+ final Tablet tablet = new Tablet(
+ Arrays.asList("region", "device", "temperature"),
+ Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.FLOAT)
+ );
+ for (int i = 0; i < ROW_COUNT; i++) {
+ tablet.addTimestamp(i, 1000L + i);
+ tablet.addValue(i, "region", i < 2 ? "beijing" : "shanghai");
+ tablet.addValue(i, "device", "d" + (i + 1));
+ tablet.addValue(i, "temperature", 21.5f + i);
+ }
+
+ try (ITsFileWriter writer = new
TsFileWriterBuilder().file(file).tableSchema(tableSchema).build()) {
+ writer.write(tablet);
+ }
+ return file;
+ }
+}
diff --git a/extensions-contrib/tsfile-extensions/README.md
b/extensions-contrib/tsfile-extensions/README.md
new file mode 100644
index 00000000000..8bcb381775f
--- /dev/null
+++ b/extensions-contrib/tsfile-extensions/README.md
@@ -0,0 +1,123 @@
+<!--
+ ~ 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.
+ -->
+
+# druid-tsfile-extensions
+
+Community extension that lets Druid batch ingestion read [Apache
TsFile](https://tsfile.apache.org/) files.
+
+User documentation:
[docs/development/extensions-contrib/tsfile.md](../../docs/development/extensions-contrib/tsfile.md).
+
+## Building
+
+```bash
+mvn -pl extensions-contrib/tsfile-extensions -am package
+```
+
+## Compatibility
+
+| Item | Version |
+|------|---------|
+| TsFile Java library | `org.apache.tsfile:tsfile:2.5.0` |
+| TsFile format | Files written by the TsFile 2.x table model (table schemas
in the file footer) |
+| Druid | 40.0.0 |
+
+The extension reads the table model. Files that only contain the legacy tree
model (device paths such as
+`root.sg.d1.s1` without a table schema in the footer) are not supported.
+
+## Dependencies and licenses
+
+The extension adds the following runtime dependencies, all Apache-2.0
compatible:
+
+| Dependency | License |
+|------------|---------|
+| `org.apache.tsfile:tsfile` | Apache-2.0 |
+| `org.apache.tsfile:common` | Apache-2.0 |
+| `com.github.luben:zstd-jni` | BSD-2-Clause (managed by Druid) |
+| `org.xerial.snappy:snappy-java` | Apache-2.0 (managed by Druid) |
+| `at.yawk.lz4:lz4-java` | Apache-2.0 (managed by Druid) |
+| `org.tukaani:xz` | Public domain (managed by Druid) |
+| `org.antlr:antlr4-runtime` | BSD-3-Clause (managed by Druid) |
+
+Only `org.apache.tsfile:tsfile` and `org.apache.tsfile:common` are new to
Druid; the compression and parser
+libraries are already shipped by Druid.
+
+## Sample data
+
+`examples/data/sensor.tsfile` is a three-row TsFile written by
`TsFileTestData`, and `examples/sensor-ingest.json`
+is an ingestion spec for it. The file contains one table, `sensor`:
+
+| Column | TsFile category | TsFile type |
+|--------|-----------------|-------------|
+| `region` | TAG | STRING |
+| `device` | TAG | STRING |
+| `temperature` | FIELD | FLOAT |
+| `humidity` | FIELD | DOUBLE |
+| `status` | FIELD | BOOLEAN |
+
+Expected input rows:
+
+| `Time` | `region` | `device` | `temperature` | `humidity` | `status` |
+|--------|----------|----------|---------------|------------|----------|
+| 1000 | beijing | d1 | 21.5 | 0.5 | true |
+| 2000 | beijing | d2 | 22.5 | *(null)* | false |
+| 3000 | shanghai | d3 | 23.5 | 0.7 | true |
+
+`Time` is in epoch milliseconds. The second row has no humidity value; TsFile
returns it as null and the extension
+omits it from the row.
+
+To regenerate the sample file from this module:
+
+```bash
+mvn exec:java -Dexec.classpathScope=test \
+ -Dexec.mainClass=org.apache.druid.data.input.tsfile.TsFileTestData
+```
+
+## Test plan
+
+`TsFileInputFormatTest` covers:
+
+| Test | What it checks |
+|------|----------------|
+| `testReadAllColumns` | All tag and field columns, timestamps, and a missing
(null) value are read correctly |
+| `testReadWithDefaultTimestampColumn` | The TsFile timestamp is also
available under a non-default `timestampSpec` column |
+| `testReadSelectedTableAndColumns` | `tables` and `columns` restrict the read
|
+| `testProjectionFromColumnsFilter` | The ingestion schema's column filter
restricts the physical columns read |
+| `testProjectionFromFixedDimensions` | A fixed dimension list (the EXTERN
signature) restricts the physical columns read |
+| `testColumnsFilterWinsOverFixedDimensions` | A batch ingestion filter keeps
every column it lists, even when the dimension list is narrower |
+| `testTagOnlyProjectionUsesRowCarrier` | A tag-only projection still returns
rows, via a row carrier that is not exposed |
+| `testIntervalPushdown` | An `interval` restricts the read to rows inside it |
+| `testIntervalIsEndExclusive` | A row at the end boundary of `interval` is
excluded |
+| `testEmptyIntervalReadsNothing` | An empty `interval` yields no rows |
+| `testReadCommittedSampleFile` | The committed `examples/data/sensor.tsfile`
matches the expected rows below |
+| `testTreeModelFileIsRejected` | A tree-model file without a table schema
fails with a clear error instead of reading zero rows |
+| `testTimestampAliasShadowsSameNamedColumn` | The TsFile time wins over a
TsFile column named like the `timestampSpec` column |
+| `testUnknownTable` | A missing table produces an `InvalidInputException`
naming the table |
+| `testUnknownColumn` | A missing column produces an `InvalidInputException`
naming the column |
+| `testSerde` | The input format round-trips through Jackson with `"type":
"tsfile"` |
+
+To run them:
+
+```bash
+mvn -pl extensions-contrib/tsfile-extensions -am test
-Dtest=TsFileInputFormatTest
+```
+
+## Maintenance
+
+This is a community extension. The Apache TsFile community owns maintenance
and compatibility testing, as agreed
+with the Druid community on
[[email protected]](https://lists.apache.org/thread/qmrkhobp7p07wkmcf4y4myq36k866zz0).
diff --git a/extensions-contrib/tsfile-extensions/examples/data/sensor.tsfile
b/extensions-contrib/tsfile-extensions/examples/data/sensor.tsfile
new file mode 100644
index 00000000000..16b5a65419f
Binary files /dev/null and
b/extensions-contrib/tsfile-extensions/examples/data/sensor.tsfile differ
diff --git a/extensions-contrib/tsfile-extensions/examples/sensor-ingest.json
b/extensions-contrib/tsfile-extensions/examples/sensor-ingest.json
new file mode 100644
index 00000000000..f5bb1d6bbe4
--- /dev/null
+++ b/extensions-contrib/tsfile-extensions/examples/sensor-ingest.json
@@ -0,0 +1,49 @@
+{
+ "type": "index_parallel",
+ "spec": {
+ "ioConfig": {
+ "type": "index_parallel",
+ "inputSource": {
+ "type": "local",
+ "baseDir":
"/path/to/druid/extensions-contrib/tsfile-extensions/examples/data",
+ "filter": "*.tsfile"
+ },
+ "inputFormat": {
+ "type": "tsfile",
+ "tables": ["sensor"],
+ "columns": ["region", "device", "temperature", "humidity", "status"]
+ }
+ },
+ "dataSchema": {
+ "dataSource": "sensor",
+ "timestampSpec": {
+ "column": "Time",
+ "format": "millis"
+ },
+ "dimensionsSpec": {
+ "dimensions": ["region", "device", "status"]
+ },
+ "metricsSpec": [
+ {
+ "type": "doubleSum",
+ "name": "temperature",
+ "fieldName": "temperature"
+ },
+ {
+ "type": "doubleSum",
+ "name": "humidity",
+ "fieldName": "humidity"
+ }
+ ],
+ "granularitySpec": {
+ "type": "uniform",
+ "segmentGranularity": "DAY",
+ "queryGranularity": "MINUTE",
+ "rollup": false
+ }
+ },
+ "tuningConfig": {
+ "type": "index_parallel"
+ }
+ }
+}
diff --git a/extensions-contrib/tsfile-extensions/pom.xml
b/extensions-contrib/tsfile-extensions/pom.xml
new file mode 100644
index 00000000000..83f58719219
--- /dev/null
+++ b/extensions-contrib/tsfile-extensions/pom.xml
@@ -0,0 +1,98 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+
+ <groupId>org.apache.druid.extensions.contrib</groupId>
+ <artifactId>druid-tsfile-extensions</artifactId>
+ <name>druid-tsfile-extensions</name>
+ <description>Apache TsFile input format for Druid batch
ingestion</description>
+
+ <parent>
+ <artifactId>druid</artifactId>
+ <groupId>org.apache.druid</groupId>
+ <version>40.0.0-SNAPSHOT</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.druid</groupId>
+ <artifactId>druid-processing</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tsfile</groupId>
+ <artifactId>tsfile</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.inject</groupId>
+ <artifactId>guice</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>joda-time</groupId>
+ <artifactId>joda-time</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <configuration>
+ <ignoredUnusedDeclaredDependencies>
+ <!-- Runtime provider for the JUnit 5 tests; declared so that
surefire can run them. -->
+
<ignoredUnusedDeclaredDependency>org.junit.jupiter:junit-jupiter-engine</ignoredUnusedDeclaredDependency>
+ </ignoredUnusedDeclaredDependencies>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileExtensionsModule.java
b/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileExtensionsModule.java
new file mode 100644
index 00000000000..8dc4d7413ee
--- /dev/null
+++
b/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileExtensionsModule.java
@@ -0,0 +1,48 @@
+/*
+ * 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.druid.data.input.tsfile;
+
+import com.fasterxml.jackson.databind.Module;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.inject.Binder;
+import org.apache.druid.initialization.DruidModule;
+
+import java.util.Collections;
+import java.util.List;
+
+public class TsFileExtensionsModule implements DruidModule
+{
+ @Override
+ public List<? extends Module> getJacksonModules()
+ {
+ return Collections.singletonList(
+ new SimpleModule("TsFileInputRowParserModule")
+ .registerSubtypes(
+ new NamedType(TsFileInputFormat.class, "tsfile")
+ )
+ );
+ }
+
+ @Override
+ public void configure(Binder binder)
+ {
+ }
+}
diff --git
a/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputEntityReader.java
b/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputEntityReader.java
new file mode 100644
index 00000000000..1925a056815
--- /dev/null
+++
b/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputEntityReader.java
@@ -0,0 +1,462 @@
+/*
+ * 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.druid.data.input.tsfile;
+
+import org.apache.druid.data.input.ColumnsFilter;
+import org.apache.druid.data.input.InputEntity;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.IntermediateRowParsingReader;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.CloseableIterators;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.common.parsers.ParseException;
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.read.ReadProcessException;
+import org.apache.tsfile.exception.write.NoMeasurementException;
+import org.apache.tsfile.exception.write.NoTableException;
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.read.query.dataset.ResultSet;
+import org.apache.tsfile.read.v4.ITsFileReader;
+import org.apache.tsfile.read.v4.TsFileReaderBuilder;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Reads the rows of a TsFile file with the TsFile table-model reader and
converts them into
+ * {@link MapBasedInputRow}-compatible maps.
+ * <p>
+ * TsFile stores its own timestamp, which is exposed to Druid under the name
{@value TsFileInputFormat#TIME_COLUMN_NAME}.
+ * Tag, field, and attribute columns are exposed as-is.
+ */
+public class TsFileInputEntityReader extends
IntermediateRowParsingReader<Map<String, Object>>
+{
+ private final InputRowSchema inputRowSchema;
+ private final InputEntity source;
+ @Nullable
+ private final File temporaryDirectory;
+ @Nullable
+ private final List<String> tables;
+ @Nullable
+ private final List<String> columns;
+ @Nullable
+ private final Interval interval;
+
+ TsFileInputEntityReader(
+ InputRowSchema inputRowSchema,
+ InputEntity source,
+ @Nullable File temporaryDirectory,
+ @Nullable List<String> tables,
+ @Nullable List<String> columns,
+ @Nullable Interval interval
+ )
+ {
+ this.inputRowSchema = inputRowSchema;
+ this.source = source;
+ this.temporaryDirectory = temporaryDirectory;
+ this.tables = tables;
+ this.columns = columns;
+ this.interval = interval;
+ }
+
+ @Override
+ protected CloseableIterator<Map<String, Object>> intermediateRowIterator()
throws IOException
+ {
+ final InputEntity.CleanableFile materialized =
+ source.fetch(temporaryDirectory, new
byte[InputEntity.DEFAULT_FETCH_BUFFER_SIZE]);
+ final ITsFileReader reader;
+ try {
+ reader = new TsFileReaderBuilder().file(materialized.file()).build();
+ }
+ catch (IOException | RuntimeException e) {
+ materialized.close();
+ throw e;
+ }
+
+ final List<String> tableNames;
+ try {
+ tableNames = resolveTableNames(reader);
+ }
+ catch (IOException | RuntimeException e) {
+ reader.close();
+ materialized.close();
+ throw e;
+ }
+
+ return CloseableIterators.wrap(
+ new RowIterator(reader, tableNames),
+ () -> {
+ reader.close();
+ materialized.close();
+ }
+ );
+ }
+
+ @Override
+ protected InputEntity source()
+ {
+ return source;
+ }
+
+ @Override
+ protected List<InputRow> parseInputRows(Map<String, Object> intermediateRow)
throws ParseException
+ {
+ return Collections.singletonList(MapInputRowParser.parse(inputRowSchema,
intermediateRow));
+ }
+
+ @Override
+ protected List<Map<String, Object>> toMap(Map<String, Object>
intermediateRow)
+ {
+ return Collections.singletonList(intermediateRow);
+ }
+
+ private List<String> resolveTableNames(ITsFileReader reader) throws
IOException
+ {
+ final List<TableSchema> allTables = reader.getAllTableSchema();
+ if (allTables.isEmpty()) {
+ throw InvalidInput.exception(
+ "TsFile[%s] contains no table schema. Only table-model TsFiles are
supported; tree-model files "
+ + "are not.",
+ source.getUri()
+ );
+ }
+ if (tables == null || tables.isEmpty()) {
+ return
allTables.stream().map(TableSchema::getTableName).collect(Collectors.toList());
+ }
+
+ final Set<String> available =
+ allTables.stream().map(schema ->
schema.getTableName().toLowerCase(Locale.ROOT)).collect(Collectors.toSet());
+ for (String table : tables) {
+ if (!available.contains(table.toLowerCase(Locale.ROOT))) {
+ throw InvalidInput.exception(
+ "Table[%s] does not exist in the file. Available tables: %s",
+ table,
+
allTables.stream().map(TableSchema::getTableName).collect(Collectors.toList())
+ );
+ }
+ }
+ return tables;
+ }
+
+ /**
+ * Columns to read for one table. {@code queryColumns} are passed to the
TsFile reader; only
+ * {@code outputColumns} end up in the row. The two differ when a column is
read purely to make the reader
+ * produce rows.
+ */
+ private static class TableProjection
+ {
+ private final List<String> queryColumns;
+ private final Set<String> outputColumns;
+
+ TableProjection(List<String> queryColumns, Set<String> outputColumns)
+ {
+ this.queryColumns = queryColumns;
+ this.outputColumns = outputColumns;
+ }
+ }
+
+ private TableProjection selectColumns(TableSchema tableSchema)
+ {
+ if (columns != null && !columns.isEmpty()) {
+ return new TableProjection(columns, new HashSet<>(columns));
+ }
+
+ final List<IMeasurementSchema> columnSchemas =
tableSchema.getColumnSchemas();
+ final List<ColumnCategory> columnCategories = tableSchema.getColumnTypes();
+ final List<String> physicalColumns = new ArrayList<>();
+ final Set<String> tagColumns = new HashSet<>();
+ final Set<String> supportedColumns = new HashSet<>();
+ for (int i = 0; i < columnSchemas.size(); i++) {
+ final String columnName = columnSchemas.get(i).getMeasurementName();
+ if (TsFileInputFormat.TIME_COLUMN_NAME.equalsIgnoreCase(columnName)) {
+ continue;
+ }
+ physicalColumns.add(columnName);
+ if (columnCategories.get(i) == ColumnCategory.TAG) {
+ tagColumns.add(columnName);
+ }
+ if (isSupportedType(columnSchemas.get(i).getType())) {
+ supportedColumns.add(columnName);
+ }
+ }
+
+ // The ingestion schema's filter is authoritative whenever it carries
information: it already lists the
+ // timestamp column, the dimensions, the transforms' inputs and the
aggregators' inputs. MSQ EXTERN always
+ // passes ColumnsFilter.all(), which carries none of that; there the
projection comes from the fixed
+ // dimensions, i.e. the external table's column signature.
+ final ColumnsFilter columnsFilter = inputRowSchema.getColumnsFilter();
+ List<String> selected = null;
+ if (columnsFilter != null && !columnsFilter.equals(ColumnsFilter.all())) {
+ selected =
physicalColumns.stream().filter(columnsFilter::apply).collect(Collectors.toList());
+ } else {
+ final DimensionsSpec dimensionsSpec = inputRowSchema.getDimensionsSpec();
+ if (dimensionsSpec != null && dimensionsSpec.hasFixedDimensions()) {
+ final Set<String> wanted = new
HashSet<>(dimensionsSpec.getDimensionNames());
+ final List<String> projected =
physicalColumns.stream().filter(wanted::contains).collect(Collectors.toList());
+ if (!projected.isEmpty()) {
+ selected = projected;
+ }
+ }
+ }
+ if (selected == null) {
+ // Nothing in the schema narrows the columns down: read everything that
can be converted to a row value.
+ selected =
physicalColumns.stream().filter(supportedColumns::contains).collect(Collectors.toList());
+ }
+
+ // The TsFile table reader only produces blocks while it has a measurement
column to read, so a projection of
+ // tag columns alone would yield no rows at all. Read one supported
measurement column as a row carrier in
+ // that case, without exposing it in the row.
+ final List<String> queryColumns = new ArrayList<>(selected);
+ if (selected.stream().noneMatch(column -> !tagColumns.contains(column))) {
+ physicalColumns.stream()
+ .filter(column -> !tagColumns.contains(column))
+ .filter(supportedColumns::contains)
+ .findFirst()
+ .ifPresent(queryColumns::add);
+ }
+
+ return new TableProjection(queryColumns, new HashSet<>(selected));
+ }
+
+ private static boolean isSupportedType(TSDataType dataType)
+ {
+ switch (dataType) {
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ case BOOLEAN:
+ case TEXT:
+ case STRING:
+ case DATE:
+ case TIMESTAMP:
+ case BLOB:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ /**
+ * Iterates over the rows of every selected table, keeping at most one open
{@link ResultSet} at a time.
+ */
+ private class RowIterator implements Iterator<Map<String, Object>>
+ {
+ private final ITsFileReader reader;
+ private final Iterator<String> tableIterator;
+
+ @Nullable
+ private ResultSet resultSet = null;
+ private List<String> currentColumns = null;
+ private List<TSDataType> currentTypes = null;
+ private Set<String> currentOutputColumns = null;
+ @Nullable
+ private Map<String, Object> bufferedRow = null;
+
+ RowIterator(ITsFileReader reader, List<String> tableNames)
+ {
+ this.reader = reader;
+ this.tableIterator = tableNames.iterator();
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ if (bufferedRow != null) {
+ return true;
+ }
+
+ while (true) {
+ if (resultSet != null) {
+ try {
+ if (resultSet.next()) {
+ bufferedRow = readCurrentRow();
+ return true;
+ }
+ }
+ catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ closeCurrentResultSet();
+ }
+
+ if (!tableIterator.hasNext()) {
+ closeCurrentResultSet();
+ return false;
+ }
+ openTable(tableIterator.next());
+ }
+ }
+
+ @Override
+ public Map<String, Object> next()
+ {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ final Map<String, Object> row = bufferedRow;
+ bufferedRow = null;
+ return row;
+ }
+
+ private void openTable(String tableName)
+ {
+ final long queryStartTime;
+ final long queryEndTime;
+ if (interval == null) {
+ queryStartTime = Long.MIN_VALUE;
+ queryEndTime = Long.MAX_VALUE;
+ } else {
+ if (interval.getEndMillis() <= interval.getStartMillis()) {
+ // An empty interval matches no row, so there is nothing to read.
+ return;
+ }
+ // TsFile folds both time bounds into the comparison, while a Druid
interval is end-exclusive.
+ queryStartTime = interval.getStartMillis();
+ queryEndTime = interval.getEndMillis() - 1;
+ }
+
+ final TableSchema tableSchema;
+ try {
+ tableSchema = reader.getTableSchemas(tableName)
+ .orElseThrow(
+ () -> InvalidInput.exception(
+ "Table[%s] does not exist in the file",
+ tableName
+ )
+ );
+ }
+ catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ final TableProjection projection = selectColumns(tableSchema);
+ final List<String> selectedColumns = projection.queryColumns;
+ final List<TSDataType> types = new ArrayList<>(selectedColumns.size());
+ for (String column : selectedColumns) {
+ final int columnIndex = tableSchema.findColumnIndex(column);
+ if (columnIndex < 0) {
+ throw InvalidInput.exception(
+ "Column[%s] does not exist in table[%s]",
+ column,
+ tableName
+ );
+ }
+ types.add(tableSchema.getColumnSchemas().get(columnIndex).getType());
+ }
+
+ try {
+ resultSet = reader.query(tableName, selectedColumns, queryStartTime,
queryEndTime);
+ }
+ catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ catch (ReadProcessException | NoTableException | NoMeasurementException
e) {
+ throw new RuntimeException(e);
+ }
+ currentColumns = selectedColumns;
+ currentTypes = types;
+ currentOutputColumns = projection.outputColumns;
+ }
+
+ private void closeCurrentResultSet()
+ {
+ if (resultSet != null) {
+ resultSet.close();
+ resultSet = null;
+ }
+ }
+
+ private Map<String, Object> readCurrentRow()
+ {
+ final Map<String, Object> row = new HashMap<>();
+
+ for (int i = 0; i < currentColumns.size(); i++) {
+ final String column = currentColumns.get(i);
+ final int resultSetIndex = i + 2;
+ if (currentOutputColumns.contains(column) &&
!resultSet.isNull(resultSetIndex)) {
+ row.put(column, readValue(resultSet, currentTypes.get(i),
resultSetIndex));
+ }
+ }
+
+ // Written after the measurements so that a column sharing the timestamp
column name is shadowed by it.
+ // Column index 1 of every TsFile result set is the implicit time column.
+ final long timestamp = resultSet.getLong(1);
+ row.put(TsFileInputFormat.TIME_COLUMN_NAME, timestamp);
+
+ final String timestampColumn =
inputRowSchema.getTimestampSpec().getTimestampColumn();
+ if (!TsFileInputFormat.TIME_COLUMN_NAME.equals(timestampColumn)) {
+ row.put(timestampColumn, timestamp);
+ }
+ return row;
+ }
+ }
+
+ private static Object readValue(ResultSet resultSet, TSDataType dataType,
int columnIndex)
+ {
+ switch (dataType) {
+ case INT32:
+ return resultSet.getInt(columnIndex);
+ case INT64:
+ return resultSet.getLong(columnIndex);
+ case FLOAT:
+ return resultSet.getFloat(columnIndex);
+ case DOUBLE:
+ return resultSet.getDouble(columnIndex);
+ case BOOLEAN:
+ return resultSet.getBoolean(columnIndex);
+ case TEXT:
+ case STRING:
+ return resultSet.getString(columnIndex);
+ case DATE:
+ return resultSet.getDate(columnIndex).toEpochDay();
+ case TIMESTAMP:
+ return resultSet.getLong(columnIndex);
+ case BLOB:
+ return
Base64.getEncoder().encodeToString(resultSet.getBinary(columnIndex));
+ default:
+ throw InvalidInput.exception(
+ "TsFile column[%s] has an unsupported data type[%s]",
+ resultSet.getMetadata().getColumnName(columnIndex),
+ dataType
+ );
+ }
+ }
+}
diff --git
a/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputFormat.java
b/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputFormat.java
new file mode 100644
index 00000000000..6be4119c2a6
--- /dev/null
+++
b/extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputFormat.java
@@ -0,0 +1,148 @@
+/*
+ * 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.druid.data.input.tsfile;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.data.input.InputEntity;
+import org.apache.druid.data.input.InputEntityReader;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRowSchema;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Reads Apache TsFile files into Druid. TsFile is a columnar file format for
time series data, so a single file
+ * can hold multiple tables. Rows of each table are converted into {@link
org.apache.druid.data.input.InputRow}s.
+ * <p>
+ * The timestamp of a TsFile row is returned as the {@code Time} column. The
ingestion spec must therefore use
+ * {@code "timestampSpec": {"column": "Time", "format": "millis"}} (or any
other numeric format).
+ */
+public class TsFileInputFormat implements InputFormat
+{
+ /** Name of the implicit time column of TsFile query results. */
+ public static final String TIME_COLUMN_NAME = "Time";
+
+ @Nullable
+ private final List<String> tables;
+ @Nullable
+ private final List<String> columns;
+ @Nullable
+ private final Interval interval;
+
+ @JsonCreator
+ public TsFileInputFormat(
+ @JsonProperty("tables") @Nullable List<String> tables,
+ @JsonProperty("columns") @Nullable List<String> columns,
+ @JsonProperty("interval") @Nullable Interval interval
+ )
+ {
+ this.tables = tables;
+ this.columns = columns;
+ this.interval = interval;
+ }
+
+ /**
+ * Tables to read. When null or empty, all tables in the file are read.
+ */
+ @Nullable
+ @JsonProperty("tables")
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public List<String> getTables()
+ {
+ return tables;
+ }
+
+ /**
+ * Columns to read. When null or empty, all columns of each table are read.
+ */
+ @Nullable
+ @JsonProperty("columns")
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public List<String> getColumns()
+ {
+ return columns;
+ }
+
+ /**
+ * Time range to read, in Druid's {@code [start, end)} semantics. When null,
each file is read in full.
+ * The range is pushed down to the TsFile reader, which skips chunks whose
time statistics fall outside of it.
+ */
+ @Nullable
+ @JsonProperty("interval")
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public Interval getInterval()
+ {
+ return interval;
+ }
+
+ @Override
+ public boolean isSplittable()
+ {
+ return false;
+ }
+
+ @Override
+ public InputEntityReader createReader(
+ InputRowSchema inputRowSchema,
+ InputEntity source,
+ File temporaryDirectory
+ )
+ {
+ return new TsFileInputEntityReader(inputRowSchema, source,
temporaryDirectory, tables, columns, interval);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TsFileInputFormat that = (TsFileInputFormat) o;
+ return Objects.equals(tables, that.tables)
+ && Objects.equals(columns, that.columns)
+ && Objects.equals(interval, that.interval);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(tables, columns, interval);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "TsFileInputFormat{" +
+ "tables=" + tables +
+ ", columns=" + columns +
+ ", interval=" + interval +
+ '}';
+ }
+}
diff --git
a/extensions-contrib/tsfile-extensions/src/main/resources/META-INF/services/org.apache.druid.initialization.DruidModule
b/extensions-contrib/tsfile-extensions/src/main/resources/META-INF/services/org.apache.druid.initialization.DruidModule
new file mode 100644
index 00000000000..1ca3048aab9
--- /dev/null
+++
b/extensions-contrib/tsfile-extensions/src/main/resources/META-INF/services/org.apache.druid.initialization.DruidModule
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.druid.data.input.tsfile.TsFileExtensionsModule
diff --git
a/extensions-contrib/tsfile-extensions/src/test/java/org/apache/druid/data/input/tsfile/TsFileInputFormatTest.java
b/extensions-contrib/tsfile-extensions/src/test/java/org/apache/druid/data/input/tsfile/TsFileInputFormatTest.java
new file mode 100644
index 00000000000..ccbb1d71b4d
--- /dev/null
+++
b/extensions-contrib/tsfile-extensions/src/test/java/org/apache/druid/data/input/tsfile/TsFileInputFormatTest.java
@@ -0,0 +1,367 @@
+/*
+ * 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.druid.data.input.tsfile;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.data.input.ColumnsFilter;
+import org.apache.druid.data.input.InputEntityReader;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.FileEntity;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.joda.time.Interval;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TsFileInputFormatTest
+{
+ @TempDir
+ public File temporaryDirectory;
+
+ private File sampleFile;
+
+ @BeforeEach
+ public void setUp() throws IOException, WriteProcessException
+ {
+ sampleFile = new File(temporaryDirectory, "sample.tsfile");
+ TsFileTestData.writeSampleFile(sampleFile);
+ }
+
+ @Test
+ public void testReadAllColumns() throws IOException
+ {
+ final InputRowSchema rowSchema =
buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis");
+ final List<InputRow> rows = read(new TsFileInputFormat(null, null, null),
rowSchema);
+
+ assertEquals(3, rows.size());
+
+ final InputRow first = rows.get(0);
+ assertEquals(1000L, first.getTimestampFromEpoch());
+ assertEquals(Collections.singletonList("beijing"),
first.getDimension("region"));
+ assertEquals(Collections.singletonList("d1"),
first.getDimension("device"));
+ assertEquals(21.5f, (Float) first.getRaw("temperature"), 0.0001f);
+ assertEquals(0.5d, (Double) first.getRaw("humidity"), 0.0001d);
+ assertEquals(Boolean.TRUE, first.getRaw("status"));
+
+ // humidity is missing for the second row.
+ final InputRow second = rows.get(1);
+ assertEquals(2000L, second.getTimestampFromEpoch());
+ assertNull(second.getRaw("humidity"));
+ }
+
+ @Test
+ public void testReadWithDefaultTimestampColumn() throws IOException
+ {
+ final InputRowSchema rowSchema = buildRowSchema("timestamp", "auto");
+ final List<InputRow> rows = read(new TsFileInputFormat(null, null, null),
rowSchema);
+
+ assertEquals(3, rows.size());
+ assertEquals(1000L, rows.get(0).getTimestampFromEpoch());
+ }
+
+ @Test
+ public void testReadSelectedTableAndColumns() throws IOException
+ {
+ final InputRowSchema rowSchema =
buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis");
+ final List<InputRow> rows = read(
+ new TsFileInputFormat(
+ Collections.singletonList(TsFileTestData.TABLE_NAME),
+ Arrays.asList("region", "temperature"),
+ null
+ ),
+ rowSchema
+ );
+
+ assertEquals(3, rows.size());
+ final InputRow first = rows.get(0);
+ assertEquals(Collections.singletonList("beijing"),
first.getDimension("region"));
+ assertNull(first.getRaw("device"));
+ assertEquals(21.5f, (Float) first.getRaw("temperature"), 0.0001f);
+ assertNull(first.getRaw("humidity"));
+ }
+
+ @Test
+ public void testUnknownTable()
+ {
+ final InputRowSchema rowSchema =
buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis");
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> read(new TsFileInputFormat(Collections.singletonList("unknown"),
null, null), rowSchema)
+ );
+ assertTrue(e.getMessage().contains("unknown"));
+ }
+
+ @Test
+ public void testUnknownColumn()
+ {
+ final InputRowSchema rowSchema =
buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis");
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> read(new TsFileInputFormat(null,
Collections.singletonList("unknown"), null), rowSchema)
+ );
+ assertTrue(e.getMessage().contains("unknown"));
+ }
+
+ @Test
+ public void testIntervalPushdown() throws IOException
+ {
+ final InputRowSchema rowSchema =
buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis");
+ final List<InputRow> rows = read(
+ new TsFileInputFormat(null, null, millis(1500L, 3501L)),
+ rowSchema
+ );
+
+ assertEquals(2, rows.size());
+ assertEquals(2000L, rows.get(0).getTimestampFromEpoch());
+ assertEquals(3000L, rows.get(1).getTimestampFromEpoch());
+ }
+
+ @Test
+ public void testIntervalIsEndExclusive() throws IOException
+ {
+ final InputRowSchema rowSchema =
buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis");
+ final List<InputRow> rows = read(
+ new TsFileInputFormat(null, null, millis(1000L, 2000L)),
+ rowSchema
+ );
+
+ assertEquals(1, rows.size());
+ assertEquals(1000L, rows.get(0).getTimestampFromEpoch());
+ }
+
+ @Test
+ public void testEmptyIntervalReadsNothing() throws IOException
+ {
+ final InputRowSchema rowSchema =
buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis");
+ final List<InputRow> rows = read(
+ new TsFileInputFormat(null, null, millis(2000L, 2000L)),
+ rowSchema
+ );
+
+ assertTrue(rows.isEmpty());
+ }
+
+ @Test
+ public void testTreeModelFileIsRejected() throws IOException,
WriteProcessException
+ {
+ final File file = new File(temporaryDirectory, "tree-model.tsfile");
+ TsFileTestData.writeTreeModelFile(file);
+
+ final DruidException e = assertThrows(
+ DruidException.class,
+ () -> read(
+ file,
+ new TsFileInputFormat(null, null, null),
+ buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis")
+ )
+ );
+ assertTrue(e.getMessage().contains("no table schema"));
+ }
+
+ @Test
+ public void testTimestampAliasShadowsSameNamedColumn() throws IOException,
WriteProcessException
+ {
+ final File file = new File(temporaryDirectory, "timestamp-column.tsfile");
+ TsFileTestData.writeFileWithTimestampColumn(file);
+
+ final List<InputRow> rows = read(
+ file,
+ new TsFileInputFormat(null, null, null),
+ buildRowSchema("timestamp", "millis")
+ );
+
+ assertEquals(1, rows.size());
+ // The TsFile time wins over the value of the TsFile column that happens
to be named "timestamp".
+ assertEquals(1000L, rows.get(0).getTimestampFromEpoch());
+ }
+
+ @Test
+ public void testProjectionFromColumnsFilter() throws IOException
+ {
+ final InputRowSchema rowSchema = new InputRowSchema(
+ new TimestampSpec(TsFileInputFormat.TIME_COLUMN_NAME, "millis", null),
+ new
DimensionsSpec(DimensionsSpec.getDefaultSchemas(Arrays.asList("region",
"device"))),
+ ColumnsFilter.inclusionBased(new HashSet<>(Arrays.asList("region",
"temperature")))
+ );
+ final List<InputRow> rows = read(new TsFileInputFormat(null, null, null),
rowSchema);
+
+ assertEquals(3, rows.size());
+ final InputRow first = rows.get(0);
+ assertEquals(Collections.singletonList("beijing"),
first.getDimension("region"));
+ assertEquals(21.5f, (Float) first.getRaw("temperature"), 0.0001f);
+ assertNull(first.getRaw("device"));
+ assertNull(first.getRaw("status"));
+ }
+
+ @Test
+ public void testProjectionFromFixedDimensions() throws IOException
+ {
+ // MSQ EXTERN passes ColumnsFilter.all() and puts the external table's
column signature in the dimensions.
+ final InputRowSchema rowSchema = new InputRowSchema(
+ new TimestampSpec(TsFileInputFormat.TIME_COLUMN_NAME, "millis", null),
+ new
DimensionsSpec(DimensionsSpec.getDefaultSchemas(Arrays.asList("region",
"temperature"))),
+ ColumnsFilter.all()
+ );
+ final List<InputRow> rows = read(new TsFileInputFormat(null, null, null),
rowSchema);
+
+ assertEquals(3, rows.size());
+ assertEquals(Collections.singletonList("beijing"),
rows.get(0).getDimension("region"));
+ assertEquals(21.5f, (Float) rows.get(0).getRaw("temperature"), 0.0001f);
+ assertNull(rows.get(0).getRaw("device"));
+ assertNull(rows.get(0).getRaw("humidity"));
+ assertNull(rows.get(0).getRaw("status"));
+ }
+
+ @Test
+ public void testTagOnlyProjectionUsesRowCarrier() throws IOException
+ {
+ // TsFile's table reader returns no rows at all when only tag columns are
selected, so the reader adds one
+ // measurement column as a row carrier without exposing it in the row.
+ final InputRowSchema rowSchema = new InputRowSchema(
+ new TimestampSpec(TsFileInputFormat.TIME_COLUMN_NAME, "millis", null),
+ new
DimensionsSpec(DimensionsSpec.getDefaultSchemas(Collections.singletonList("region"))),
+ ColumnsFilter.all()
+ );
+ final List<InputRow> rows = read(new TsFileInputFormat(null, null, null),
rowSchema);
+
+ assertEquals(3, rows.size());
+ assertEquals(Collections.singletonList("beijing"),
rows.get(0).getDimension("region"));
+ assertNull(rows.get(0).getRaw("temperature"));
+ assertNull(rows.get(0).getRaw("device"));
+ }
+
+ @Test
+ public void testColumnsFilterWinsOverFixedDimensions() throws IOException
+ {
+ // A batch ingestion filter that happens to select every physical column
must not be narrowed down to the
+ // dimensions: its extra entries are transform inputs or aggregator inputs.
+ final InputRowSchema rowSchema = new InputRowSchema(
+ new TimestampSpec(TsFileInputFormat.TIME_COLUMN_NAME, "millis", null),
+ new
DimensionsSpec(DimensionsSpec.getDefaultSchemas(Collections.singletonList("region"))),
+ ColumnsFilter.inclusionBased(
+ new HashSet<>(Arrays.asList("region", "device", "temperature",
"humidity", "status"))
+ )
+ );
+ final List<InputRow> rows = read(new TsFileInputFormat(null, null, null),
rowSchema);
+
+ assertEquals(3, rows.size());
+ assertEquals(21.5f, (Float) rows.get(0).getRaw("temperature"), 0.0001f);
+ assertEquals(0.5d, (Double) rows.get(0).getRaw("humidity"), 0.0001d);
+ assertEquals(Boolean.TRUE, rows.get(0).getRaw("status"));
+ }
+
+ @Test
+ public void testSerde() throws IOException
+ {
+ final ObjectMapper mapper = new DefaultObjectMapper();
+ new
TsFileExtensionsModule().getJacksonModules().forEach(mapper::registerModule);
+
+ final String json =
"{\"type\":\"tsfile\",\"tables\":[\"sensor\"],\"columns\":[\"region\",\"temperature\"],"
+ +
"\"interval\":\"1970-01-01T00:00:00.000Z/1970-01-01T00:00:04.000Z\"}";
+ final TsFileInputFormat format = (TsFileInputFormat)
mapper.readValue(json, InputFormat.class);
+ assertEquals(Collections.singletonList(TsFileTestData.TABLE_NAME),
format.getTables());
+ assertEquals(Arrays.asList("region", "temperature"), format.getColumns());
+ assertEquals(millis(0L, 4000L), format.getInterval());
+ assertEquals(format, mapper.readValue(mapper.writeValueAsString(format),
InputFormat.class));
+ }
+
+ @Test
+ public void testReadCommittedSampleFile() throws IOException
+ {
+ final File file = new File("examples/data/sensor.tsfile");
+ final List<InputRow> rows = read(
+ file,
+ new TsFileInputFormat(null, null, null),
+ buildRowSchema(TsFileInputFormat.TIME_COLUMN_NAME, "millis")
+ );
+
+ assertEquals(TsFileTestData.TIMESTAMPS.length, rows.size());
+ for (int i = 0; i < rows.size(); i++) {
+ final InputRow row = rows.get(i);
+ assertEquals(TsFileTestData.TIMESTAMPS[i], row.getTimestampFromEpoch());
+ assertEquals(Collections.singletonList(TsFileTestData.REGIONS[i]),
row.getDimension("region"));
+ assertEquals(Collections.singletonList(TsFileTestData.DEVICES[i]),
row.getDimension("device"));
+ assertEquals(TsFileTestData.TEMPERATURES[i], (Float)
row.getRaw("temperature"), 0.0001f);
+ if (Double.isNaN(TsFileTestData.HUMIDITIES[i])) {
+ assertNull(row.getRaw("humidity"));
+ } else {
+ assertEquals(TsFileTestData.HUMIDITIES[i], (Double)
row.getRaw("humidity"), 0.0001d);
+ }
+ }
+ }
+
+ private List<InputRow> read(TsFileInputFormat format, InputRowSchema
rowSchema) throws IOException
+ {
+ return read(sampleFile, format, rowSchema);
+ }
+
+ private List<InputRow> read(File file, TsFileInputFormat format,
InputRowSchema rowSchema) throws IOException
+ {
+ final InputEntityReader reader = format.createReader(rowSchema, new
FileEntity(file), temporaryDirectory);
+ final List<InputRow> rows = new ArrayList<>();
+ try (CloseableIterator<InputRow> iterator = reader.read()) {
+ while (iterator.hasNext()) {
+ rows.add(iterator.next());
+ }
+ }
+ return rows;
+ }
+
+ private static Interval millis(long startMillis, long endMillis)
+ {
+ return new Interval(DateTimes.utc(startMillis), DateTimes.utc(endMillis));
+ }
+
+ /**
+ * Schema used by most tests: no column projection is applied, so the TsFile
columns are read in full.
+ */
+ private static InputRowSchema buildRowSchema(String timestampColumn, String
timestampFormat)
+ {
+ return new InputRowSchema(
+ new TimestampSpec(timestampColumn, timestampFormat, null),
+ DimensionsSpec.builder()
+
.setDimensions(DimensionsSpec.getDefaultSchemas(Arrays.asList("region",
"device")))
+ .useSchemaDiscovery(true)
+ .build(),
+ ColumnsFilter.all()
+ );
+ }
+}
diff --git
a/extensions-contrib/tsfile-extensions/src/test/java/org/apache/druid/data/input/tsfile/TsFileTestData.java
b/extensions-contrib/tsfile-extensions/src/test/java/org/apache/druid/data/input/tsfile/TsFileTestData.java
new file mode 100644
index 00000000000..4d4c774b1ee
--- /dev/null
+++
b/extensions-contrib/tsfile-extensions/src/test/java/org/apache/druid/data/input/tsfile/TsFileTestData.java
@@ -0,0 +1,149 @@
+/*
+ * 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.druid.data.input.tsfile;
+
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.file.metadata.ColumnSchemaBuilder;
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.record.TSRecord;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.record.datapoint.IntDataPoint;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.apache.tsfile.write.v4.ITsFileWriter;
+import org.apache.tsfile.write.v4.TsFileWriterBuilder;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+
+/**
+ * Writes the sample TsFile used by {@link TsFileInputFormatTest} and by the
example ingestion spec in
+ * {@code examples/}. Also serves as documentation for the TsFile layout the
extension reads.
+ */
+public class TsFileTestData
+{
+ public static final String TABLE_NAME = "sensor";
+ public static final long[] TIMESTAMPS = {1000L, 2000L, 3000L};
+ public static final String[] REGIONS = {"beijing", "beijing", "shanghai"};
+ public static final String[] DEVICES = {"d1", "d2", "d3"};
+ public static final float[] TEMPERATURES = {21.5f, 22.5f, 23.5f};
+ /** NaN marks a row where the humidity is not written, which TsFile reads
back as null. */
+ public static final double[] HUMIDITIES = {0.5d, Double.NaN, 0.7d};
+
+ private TsFileTestData()
+ {
+ }
+
+ /**
+ * Writes {@code examples/data/sensor.tsfile}, used by the example ingestion
spec. Run from this module:
+ * <pre>
+ * mvn exec:java -Dexec.classpathScope=test \
+ * -Dexec.mainClass=org.apache.druid.data.input.tsfile.TsFileTestData
+ * </pre>
+ */
+ public static void main(String[] args) throws IOException,
WriteProcessException
+ {
+ final Path outputDir = Paths.get(args.length > 0 ? args[0] :
"examples/data");
+ Files.createDirectories(outputDir);
+ final File file = outputDir.resolve("sensor.tsfile").toFile();
+ writeSampleFile(file);
+ System.out.println("Wrote " + file.getAbsolutePath());
+ }
+
+ public static void writeSampleFile(File file) throws IOException,
WriteProcessException
+ {
+ final TableSchema tableSchema = new TableSchema(
+ TABLE_NAME,
+ Arrays.asList(
+ new
ColumnSchemaBuilder().name("region").dataType(TSDataType.STRING).category(ColumnCategory.TAG).build(),
+ new
ColumnSchemaBuilder().name("device").dataType(TSDataType.STRING).category(ColumnCategory.TAG).build(),
+ new
ColumnSchemaBuilder().name("temperature").dataType(TSDataType.FLOAT).build(),
+ new
ColumnSchemaBuilder().name("humidity").dataType(TSDataType.DOUBLE).build(),
+ new
ColumnSchemaBuilder().name("status").dataType(TSDataType.BOOLEAN).build()
+ )
+ );
+
+ final Tablet tablet = new Tablet(
+ Arrays.asList("region", "device", "temperature", "humidity", "status"),
+ Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.FLOAT,
TSDataType.DOUBLE, TSDataType.BOOLEAN)
+ );
+
+ for (int i = 0; i < TIMESTAMPS.length; i++) {
+ tablet.addTimestamp(i, TIMESTAMPS[i]);
+ tablet.addValue(i, "region", REGIONS[i]);
+ tablet.addValue(i, "device", DEVICES[i]);
+ tablet.addValue(i, "temperature", TEMPERATURES[i]);
+ tablet.addValue(i, "status", i % 2 == 0);
+ if (!Double.isNaN(HUMIDITIES[i])) {
+ tablet.addValue(i, "humidity", HUMIDITIES[i]);
+ }
+ }
+
+ try (ITsFileWriter writer = new
TsFileWriterBuilder().file(file).tableSchema(tableSchema).build()) {
+ writer.write(tablet);
+ }
+ }
+
+ /**
+ * Writes a legacy tree-model TsFile, which has no table schema in its
footer and is not readable by the
+ * extension.
+ */
+ public static void writeTreeModelFile(File file) throws IOException,
WriteProcessException
+ {
+ try (TsFileWriter writer = new TsFileWriter(file)) {
+ writer.registerTimeseries("root.sg.d1", new MeasurementSchema("s1",
TSDataType.INT32));
+ writer.writeRecord(new TSRecord("root.sg.d1",
TIMESTAMPS[0]).addTuple(new IntDataPoint("s1", 1)));
+ writer.flush();
+ }
+ }
+
+ /**
+ * Writes a table-model file that has a value column literally named {@code
timestamp}, so that tests can check
+ * the timestamp alias wins over a same-named column.
+ */
+ public static void writeFileWithTimestampColumn(File file) throws
IOException, WriteProcessException
+ {
+ final TableSchema tableSchema = new TableSchema(
+ TABLE_NAME,
+ Arrays.asList(
+ new
ColumnSchemaBuilder().name("timestamp").dataType(TSDataType.INT64).build(),
+ new
ColumnSchemaBuilder().name("temperature").dataType(TSDataType.FLOAT).build()
+ )
+ );
+
+ final Tablet tablet = new Tablet(
+ Arrays.asList("timestamp", "temperature"),
+ Arrays.asList(TSDataType.INT64, TSDataType.FLOAT)
+ );
+ tablet.addTimestamp(0, TIMESTAMPS[0]);
+ tablet.addValue(0, "timestamp", 999_999L);
+ tablet.addValue(0, "temperature", TEMPERATURES[0]);
+
+ try (ITsFileWriter writer = new
TsFileWriterBuilder().file(file).tableSchema(tableSchema).build()) {
+ writer.write(tablet);
+ }
+ }
+}
diff --git a/licenses.yaml b/licenses.yaml
index 40cfad635cd..ea84a8066e1 100644
--- a/licenses.yaml
+++ b/licenses.yaml
@@ -7686,3 +7686,20 @@ module: extensions/druid-protobuf-extensions
license_name: Apache License version 2.0
libraries:
- org.apache.kafka: kafka-clients
+
+---
+
+name: Apache TsFile
+license_category: binary
+version: 2.5.0
+module: extensions-contrib/druid-tsfile-extensions
+license_name: Apache License version 2.0
+libraries:
+ - org.apache.tsfile: tsfile
+ - org.apache.tsfile: common
+notice: |
+ TsFile: Java: TsFile
+ Copyright 2024 The Apache Software Foundation
+
+ This product includes software developed at
+ The Apache Software Foundation (http://www.apache.org/).
diff --git a/pom.xml b/pom.xml
index 58874723b55..c65c9201d67 100644
--- a/pom.xml
+++ b/pom.xml
@@ -85,6 +85,7 @@
<avatica.version>1.29.0</avatica.version>
<avro.version>1.12.2</avro.version>
<parquet.version>1.18.1</parquet.version>
+ <tsfile.version>2.5.0</tsfile.version>
<bytebuddy.version>1.18.15</bytebuddy.version>
<!--
The base calcite parser was copied into the project; when updating
Calcite run dev/upgrade-calcite-parser to adopt upstream changes
@@ -270,6 +271,7 @@
<module>extensions-contrib/spectator-histogram</module>
<module>extensions-contrib/rabbit-stream-indexing-service</module>
<module>extensions-contrib/druid-ranger-security</module>
+ <module>extensions-contrib/tsfile-extensions</module>
<!-- distribution packaging -->
<module>distribution</module>
<!-- more test stuff -->
@@ -593,6 +595,11 @@
<artifactId>zookeeper-jute</artifactId>
<version>${zookeeper.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.tsfile</groupId>
+ <artifactId>tsfile</artifactId>
+ <version>${tsfile.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-client</artifactId>
@@ -2344,6 +2351,7 @@
<exclude>**/*.json</exclude>
<exclude>**/*.parq</exclude>
<exclude>**/*.parquet</exclude>
+ <exclude>**/*.tsfile</exclude>
<exclude>**/*.pmd</exclude> <!-- Artifact of
maven-pmd-plugin -->
<exclude>**/*.pmdruleset.xml</exclude> <!--
Artifact of maven-pmd-plugin -->
<exclude>**/docker/schema-registry/*</exclude>
diff --git a/website/.spelling b/website/.spelling
index 6b5a8c47d63..3ce16dc0927 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -2711,3 +2711,13 @@ getters
savepoints
Scrollable
updatable
+
+- ../docs/development/extensions-contrib/tsfile.md
+TsFile
+IoTDB
+INT32
+INT64
+
+- ../docs/configuration/extensions.md
+druid-tsfile-extensions
+TsFile
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
