Hi all,
Xuan Wang suggested I look at extending Flink to support the table model.
Before
writing any code I would like to agree on the shape, so this is a proposal
rather than a patch.
Up front: I have not run either Flink connector. Everything below comes from
reading the source in iotdb-extras, and I have tried to make each claim
checkable so that anyone can disagree with a file reference rather than
with me.
This is a model problem, not a configuration one
------------------------------------------------
My first assumption was that table-model support meant adding a dialect
option
and a database name. It does not. The schema mapping in
flink-sql-iotdb-connector *is* the tree model.
In IoTDBSinkFunction, a Flink column name is parsed as an IoTDB path and
split
into a device and a measurement:
:132-136 PathUtils.splitPathToDetachedNodes(fieldName);
measurement = nodes[nodes.length - 1];
device = join(copyOfRange(nodes, 0, nodes.length - 1), '.');
:108-113 session.insertAlignedRecord(...) / session.insertRecord(...)
:86 new Session.Builder().nodeUrls(..).username(..).password(..).build()
So a user must name their Flink columns as full IoTDB paths, and the last
dot
decides where the device ends. In table mode there is no path to split: a
column
is a TAG, FIELD or ATTRIBUTE under database.table, and which of the three
it is
carries meaning that a name cannot express. The tree model is not an unset
option here, it is the mapping.
The option list agrees. flink-sql-iotdb-connector currently declares:
nodeUrls mode
user sql
password aligned
cdc.port cdc.mode
cdc.pattern cdc.task[.]name
lookup.cache.max-rows lookup.cache.ttl-sec
scan.bounded.lower-bound scan.bounded.upper-bound
There is no database and no dialect, and `aligned` and `cdc.pattern` are
tree
concepts. (The bracket above is only to stop a mail client turning that
option
into a hyperlink.)
The shape: a separate module, as the project already chose for Spark
--------------------------------------------------------------------
connectors/ already contains both spark-iotdb-connector and
spark-iotdb-table-connector, and the table one is not a variant of the tree
one.
It is a parallel tree: its own parent pom, its own spark-iotdb-table-common
shared layer, and per-Spark-version modules iotdb-table-connector-3.3, -3.4
and
-3.5. Its pom does not reference spark-iotdb-connector at all.
The obvious objection to a separate Flink module is duplication of the
existing
CDC, lookup and bounded-scan machinery. That objection applies equally to
the
Spark split, and the project accepted it there, including building a second
-common layer rather than sharing the first. So the cost is one this
repository
has already weighed for this exact problem.
I would propose the same shape: a new flink-iotdb-table-connector, with
flink-sql-iotdb-connector left alone.
Why not extend the existing connector
--------------------------------------
The alternative is a dialect option plus a second mapping path inside
flink-sql-iotdb-connector. I do not think that survives contact with the
option
list: `aligned` and `cdc.pattern` have no meaning in table mode, so setting
them
would be silently ignored rather than rejected.
That is the same failure shape I raised on apache/iotdb#17439, where a
deprecated
overload keeps a parameter it no longer honours, so a caller who passes it
gets
different behaviour with no error instead of a compile failure. I argued
there
that it should be removed rather than left inert, and I would rather not
propose
the same trap here.
What I am asking
----------------
Agreement on the shape, not a schedule. Specifically:
1. Does a separate flink-iotdb-table-connector match how the project wants
this
done, given the Spark precedent, or is there a reason Flink should differ?
2. I would suggest starting with the sink, because the two sides look
structurally different rather than merely different in size. The sink
carries
the schema mapping above. The bounded source does not appear to map
anything:
IoTDBBoundedScanFunction takes the user's own `sql` option and wraps it,
:107-111 String.format("%s WHERE TIME <= %d", this.sql, upperBound)
:114 session.executeQueryStatement(sql)
and neither SchemaWrapper nor the row handling in Utils splits a path --
Utils reads rowRecord.getTimestamp() and getFields() generically. So the
source's tree couplings look like the dialect-less Session at :88 and the
TIME clause, which is a smaller and different problem from a mapping that
has
no table-mode analogue at all. If that reading is right the sink is where
the
design work actually is. I have not read the CDC or lookup paths, so please
correct me if the source has couplings I have not found.
3. Is anyone already working on this? I searched issues and pull requests in
both iotdb-extras and iotdb and found nothing on Flink and the table model,
but a search is not the same as asking.
If the direction is agreed I will open an issue with a concrete design and
take
it from there.
Best regards,
Zihan Dai
GitHub: PDGGK
