yunfengzhou-hub commented on code in PR #10833:
URL: https://github.com/apache/paimon/pull/10833#discussion_r4409814250
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java:
##########
@@ -81,12 +82,74 @@ public abstract class AbstractFlinkTableFactory
@Nullable private final FlinkCatalog flinkCatalog;
+ /** Builds the source and sink of a lake-stream table; {@code null} when
no store is named. */
+ @Nullable private final LakeStreamTableFactory lakeStreamTableFactory;
+
public AbstractFlinkTableFactory(@Nullable FlinkCatalog flinkCatalog) {
+ this(flinkCatalog, null);
+ }
+
+ public AbstractFlinkTableFactory(
+ @Nullable FlinkCatalog flinkCatalog,
+ @Nullable LakeStreamTableFactory lakeStreamTableFactory) {
this.flinkCatalog = flinkCatalog;
+ this.lakeStreamTableFactory = lakeStreamTableFactory;
}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
+ if (isLakeStreamRead(context)) {
+ return createLakeStreamSource(context);
+ }
+ return createPaimonTableSource(context);
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ if (isLakeStreamWrite(context)) {
+ return createLakeStreamSink(context);
+ }
+ return createPaimonTableSink(context, buildPaimonTable(context));
Review Comment:
Fixed. The write now reads lakestream.enabled from the schema persisted at
path — the CatalogContext carries only path and branch, so connector options
cannot restate it — and the read side is refused too: a $rt read or write
through a connector mapping fails by name instead of being answered from
Paimon, which a $rt read previously was. A plain-name read still returns Paimon
snapshots, which is what the original name means on every path.
Covered by factory-level tests and an end-to-end case with a temporary table
pointing at the table's directory, and the Flink page documents that a
lake-stream table is read and written through the catalog.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
