[flink] Union read decouple with paimon for log table (#1527)

Author: caozhen1937

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/catalog/FlinkCatalog.java

@@ -24,7 +24,7 @@
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.exception.FlussRuntimeException;
 import com.alibaba.fluss.exception.InvalidTableException;
-import com.alibaba.fluss.flink.lakehouse.LakeCatalog;
+import com.alibaba.fluss.flink.lake.LakeCatalog;
 import com.alibaba.fluss.flink.procedure.ProcedureManager;
 import com.alibaba.fluss.flink.utils.CatalogExceptionUtils;
 import com.alibaba.fluss.flink.utils.DataLakeUtils;

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/catalog/FlinkTableFactory.java

@@ -20,7 +20,7 @@
 import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.flink.FlinkConnectorOptions;
-import com.alibaba.fluss.flink.lakehouse.LakeTableFactory;
+import com.alibaba.fluss.flink.lake.LakeTableFactory;
 import com.alibaba.fluss.flink.sink.FlinkTableSink;
 import com.alibaba.fluss.flink.source.FlinkTableSource;
 import com.alibaba.fluss.flink.utils.FlinkConnectorOptionsUtils;
@@ -143,7 +143,8 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 cache,
                 partitionDiscoveryIntervalMs,
                 tableOptions.get(toFlinkOption(ConfigOptions.TABLE_DATALAKE_ENABLED)),
-                tableOptions.get(toFlinkOption(ConfigOptions.TABLE_MERGE_ENGINE)));
+                tableOptions.get(toFlinkOption(ConfigOptions.TABLE_MERGE_ENGINE)),
+                context.getCatalogTable().getOptions());
     }
 
     @Override

…a/fluss/flink/lakehouse/LakeCatalog.java …libaba/fluss/flink/lake/LakeCatalog.javafluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lakehouse/LakeCatalog.java renamed to fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lake/LakeCatalog.java fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lakehouse/LakeCatalog.java renamed to fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lake/LakeCatalog.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package com.alibaba.fluss.flink.lakehouse;
+package com.alibaba.fluss.flink.lake;
 
 import org.apache.flink.table.catalog.CatalogBaseTable;
 import org.apache.flink.table.catalog.ObjectPath;

…k/lakehouse/LakeRecordRecordEmitter.java …/flink/lake/LakeRecordRecordEmitter.javafluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lakehouse/LakeRecordRecordEmitter.java renamed to fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lake/LakeRecordRecordEmitter.java fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lakehouse/LakeRecordRecordEmitter.java renamed to fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lake/LakeRecordRecordEmitter.java

@@ -15,9 +15,10 @@
  * limitations under the License.
  */
 
-package com.alibaba.fluss.flink.lakehouse;
+package com.alibaba.fluss.flink.lake;
 
 import com.alibaba.fluss.client.table.scanner.ScanRecord;
+import com.alibaba.fluss.flink.lake.state.LakeSnapshotSplitState;
 import com.alibaba.fluss.flink.lakehouse.paimon.split.PaimonSnapshotAndFlussLogSplitState;
 import com.alibaba.fluss.flink.lakehouse.paimon.split.PaimonSnapshotSplitState;
 import com.alibaba.fluss.flink.source.reader.RecordAndPos;
@@ -48,6 +49,9 @@ public void emitRecord(
             ((PaimonSnapshotAndFlussLogSplitState) splitState)
                     .setRecordsToSkip(recordAndPos.readRecordsCount());
             sourceOutputFunc.accept(recordAndPos.record(), sourceOutput);
+        } else if (splitState instanceof LakeSnapshotSplitState) {
+            ((LakeSnapshotSplitState) splitState).setRecordsToSkip(recordAndPos.readRecordsCount());
+            sourceOutputFunc.accept(recordAndPos.record(), sourceOutput);
         } else {
             throw new UnsupportedOperationException(
                     "Unknown split state type: " + splitState.getClass());