This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch flink-iotdb-table-connector
in repository https://gitbox.apache.org/repos/asf/iotdb-extras.git
The following commit(s) were added to refs/heads/flink-iotdb-table-connector by
this push:
new 9e5206f fix: apply CDC projection in iterator and reject CDC pushdown
9e5206f is described below
commit 9e5206f98604b040de208581036b97d436549efc
Author: shuwenwei <[email protected]>
AuthorDate: Wed Sep 23 16:27:05 2026 +0800
fix: apply CDC projection in iterator and reject CDC pushdown
Move projection into SubscriptionDataIterator so deserializers read output
positions, and reject filter/aggregate/limit pushdown for the CDC source.
---
.../flink/source/cdc/SubscriptionDataIterator.java | 66 +++++++++++++++++-----
.../flink/source/common/IoTDBDataIterator.java | 5 ++
.../flink/source/cdc/IoTDBCDCSource.java | 11 +++-
.../source/cdc/IoTDBSubscriptionSourceReader.java | 6 +-
.../table/IoTDBRelationalDynamicTableSource.java | 16 +++++-
5 files changed, 87 insertions(+), 17 deletions(-)
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java
index 1b1b785..bee63a7 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/cdc/SubscriptionDataIterator.java
@@ -25,15 +25,55 @@ import
org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler.S
import java.io.IOException;
import java.sql.Timestamp;
import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
+import java.util.Map;
-/** Adapts a subscription {@link SubscriptionResultSet} to the unified {@link
IoTDBDataIterator}. */
+/**
+ * Adapts a subscription {@link SubscriptionResultSet} to the unified {@link
IoTDBDataIterator}.
+ *
+ * <p>Projection is applied here: the iterator exposes only the projected
columns, in the projected
+ * order, and maps them onto the full subscription record. Deserializers can
therefore read the
+ * columns by their output position without knowing about the projection.
+ */
public class SubscriptionDataIterator implements IoTDBDataIterator {
private final SubscriptionResultSet resultSet;
+ private final List<String> columnNames;
+ private final int[] sourceIndexes;
- public SubscriptionDataIterator(SubscriptionResultSet resultSet) {
+ public SubscriptionDataIterator(
+ SubscriptionResultSet resultSet, List<String> projectedColumns) {
this.resultSet = resultSet;
+ this.columnNames = Collections.unmodifiableList(new
ArrayList<>(projectedColumns));
+ this.sourceIndexes = resolveSourceIndexes(resultSet.getColumnNames(),
projectedColumns);
+ }
+
+ private static int[] resolveSourceIndexes(
+ List<String> availableColumns, List<String> projectedColumns) {
+ Map<String, Integer> indexByName = new HashMap<>(availableColumns.size());
+ for (int i = 0; i < availableColumns.size(); i++) {
+ indexByName.put(normalize(availableColumns.get(i)), i);
+ }
+ int[] indexes = new int[projectedColumns.size()];
+ for (int i = 0; i < projectedColumns.size(); i++) {
+ Integer index = indexByName.get(normalize(projectedColumns.get(i)));
+ if (index == null) {
+ throw new IllegalArgumentException(
+ "Projected column '"
+ + projectedColumns.get(i)
+ + "' does not exist in the IoTDB result set.");
+ }
+ indexes[i] = index;
+ }
+ return indexes;
+ }
+
+ private static String normalize(String columnName) {
+ return columnName.trim().toLowerCase(Locale.ROOT);
}
@Override
@@ -43,56 +83,56 @@ public class SubscriptionDataIterator implements
IoTDBDataIterator {
@Override
public List<String> getColumnNames() {
- return resultSet.getColumnNames();
+ return columnNames;
}
@Override
public boolean isNull(int columnIndex) {
- return resultSet.isNull(columnIndex);
+ return resultSet.isNull(sourceIndexes[columnIndex]);
}
@Override
public boolean getBoolean(int columnIndex) {
- return resultSet.getBoolean(columnIndex);
+ return resultSet.getBoolean(sourceIndexes[columnIndex]);
}
@Override
public int getInt(int columnIndex) {
- return resultSet.getInt(columnIndex);
+ return resultSet.getInt(sourceIndexes[columnIndex]);
}
@Override
public long getLong(int columnIndex) {
- return resultSet.getLong(columnIndex);
+ return resultSet.getLong(sourceIndexes[columnIndex]);
}
@Override
public float getFloat(int columnIndex) {
- return resultSet.getFloat(columnIndex);
+ return resultSet.getFloat(sourceIndexes[columnIndex]);
}
@Override
public double getDouble(int columnIndex) {
- return resultSet.getDouble(columnIndex);
+ return resultSet.getDouble(sourceIndexes[columnIndex]);
}
@Override
public String getString(int columnIndex) {
- return resultSet.getString(columnIndex);
+ return resultSet.getString(sourceIndexes[columnIndex]);
}
@Override
public byte[] getBinary(int columnIndex) {
- return resultSet.getBinary(columnIndex);
+ return resultSet.getBinary(sourceIndexes[columnIndex]);
}
@Override
public LocalDate getDate(int columnIndex) {
- return resultSet.getDate(columnIndex);
+ return resultSet.getDate(sourceIndexes[columnIndex]);
}
@Override
public Timestamp getTimestamp(int columnIndex) {
- return new Timestamp(resultSet.getLong(columnIndex));
+ return new Timestamp(resultSet.getLong(sourceIndexes[columnIndex]));
}
}
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDataIterator.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDataIterator.java
index c334dd8..1515cb2 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDataIterator.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-base/src/main/java/org/apache/iotdb/relational/flink/source/common/IoTDBDataIterator.java
@@ -29,6 +29,11 @@ import java.util.List;
*
* <p>The caller advances with {@link #next()} and then reads the current row
by column index. The
* index is always 0-based, regardless of the underlying IoTDB API.
+ *
+ * <p>The exposed columns are the final output columns of the source:
projection has already been
+ * applied. Implementations are responsible for hiding the projection (e.g.
the CDC iterator maps
+ * the projected positions onto the full subscription record), so a {@link
+ * IoTDBDeserializationSchema} can simply read the columns by their output
position.
*/
public interface IoTDBDataIterator {
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBCDCSource.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBCDCSource.java
index 379b216..50df175 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBCDCSource.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBCDCSource.java
@@ -34,6 +34,7 @@ import
org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.table.types.DataType;
/**
* Unbounded CDC source backed by the IoTDB subscription API.
@@ -48,10 +49,15 @@ public class IoTDBCDCSource<OUT>
private static final long serialVersionUID = 1L;
private final IoTDBOptions options;
+ private final DataType projectedRowType;
private final IoTDBDeserializationSchema<OUT> deserializer;
- public IoTDBCDCSource(IoTDBOptions options, IoTDBDeserializationSchema<OUT>
deserializer) {
+ public IoTDBCDCSource(
+ IoTDBOptions options,
+ DataType projectedRowType,
+ IoTDBDeserializationSchema<OUT> deserializer) {
this.options = options;
+ this.projectedRowType = projectedRowType;
this.deserializer = deserializer;
}
@@ -62,7 +68,8 @@ public class IoTDBCDCSource<OUT>
@Override
public SourceReader<OUT, IoTDBSubscriptionSplit>
createReader(SourceReaderContext readerContext) {
- return new IoTDBSubscriptionSourceReader<>(readerContext, options,
deserializer);
+ return new IoTDBSubscriptionSourceReader<>(
+ readerContext, options, projectedRowType, deserializer);
}
@Override
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java
index a668bc0..878759d 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/source/cdc/IoTDBSubscriptionSourceReader.java
@@ -32,6 +32,7 @@ import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.table.types.DataType;
import org.apache.tsfile.read.query.dataset.ResultSet;
import java.time.Duration;
@@ -52,6 +53,7 @@ public class IoTDBSubscriptionSourceReader<OUT>
private final SourceReaderContext context;
private final IoTDBOptions options;
+ private final List<String> projectedColumns;
private final IoTDBDeserializationSchema<OUT> deserializer;
private final Deque<IoTDBSubscriptionSplit> pendingSplits = new
ArrayDeque<>();
private final Deque<OUT> buffer = new ArrayDeque<>();
@@ -65,9 +67,11 @@ public class IoTDBSubscriptionSourceReader<OUT>
public IoTDBSubscriptionSourceReader(
SourceReaderContext context,
IoTDBOptions options,
+ DataType projectedRowType,
IoTDBDeserializationSchema<OUT> deserializer) {
this.context = context;
this.options = options;
+ this.projectedColumns = DataType.getFieldNames(projectedRowType);
this.deserializer = deserializer;
}
@@ -172,7 +176,7 @@ public class IoTDBSubscriptionSourceReader<OUT>
}
for (ResultSet resultSet : message.getResultSets()) {
SubscriptionDataIterator iterator =
- new SubscriptionDataIterator((SubscriptionResultSet) resultSet);
+ new SubscriptionDataIterator((SubscriptionResultSet) resultSet,
projectedColumns);
while (iterator.next()) {
OUT record = deserializer.deserialize(iterator);
if (record != null) {
diff --git
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
index 966651f..b7cbaac 100644
---
a/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
+++
b/connectors/flink-iotdb-table-connector/flink-iotdb-table-connector-flink1/src/main/java/org/apache/iotdb/relational/flink/table/IoTDBRelationalDynamicTableSource.java
@@ -104,7 +104,7 @@ public class IoTDBRelationalDynamicTableSource
if (options.isCdc()) {
return SourceProvider.of(
new IoTDBCDCSource<RowData>(
- options, new RowDataDeserializationSchema(physicalRowDataType)));
+ options, physicalRowDataType, new
RowDataDeserializationSchema(physicalRowDataType)));
}
return SourceProvider.of(
new IoTDBSource<>(
@@ -172,6 +172,11 @@ public class IoTDBRelationalDynamicTableSource
return Result.of(Collections.emptyList(), Collections.emptyList());
}
+ // The CDC source reads the subscription stream and cannot apply row-level
filters.
+ if (options.isCdc()) {
+ return Result.of(Collections.emptyList(), filters);
+ }
+
List<ResolvedExpression> acceptedFilters = new ArrayList<>();
List<ResolvedExpression> remainingFilters = new ArrayList<>();
IoTDBExpressionVisitor expressionVisitor = new IoTDBExpressionVisitor();
@@ -192,6 +197,10 @@ public class IoTDBRelationalDynamicTableSource
List<int[]> groupingSets,
List<AggregateExpression> aggregateExpressions,
DataType producedDataType) {
+ // The CDC source cannot aggregate an unbounded subscription stream.
+ if (options.isCdc()) {
+ return false;
+ }
// Grouping and argument indices refer to the scan's current row type,
which is the row type
// after any projection that has already been pushed into this source.
AggregateSpec spec =
@@ -207,6 +216,11 @@ public class IoTDBRelationalDynamicTableSource
@Override
public void applyLimit(long limit) {
+ // Limit pushdown is only an optimization hint; the CDC source ignores it
and Flink still
+ // enforces the limit itself.
+ if (options.isCdc()) {
+ return;
+ }
if (aggregateSpec == null) {
this.limit = limit;
}
