[lake] Pass Fluss schema to lake writer (#1192)

Author: xx789633

fluss-common/src/main/java/com/alibaba/fluss/lake/writer/WriterInitContext.java

@@ -18,6 +18,7 @@
 package com.alibaba.fluss.lake.writer;
 
 import com.alibaba.fluss.annotation.PublicEvolving;
+import com.alibaba.fluss.metadata.Schema;
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TablePath;
 
@@ -53,4 +54,11 @@ public interface WriterInitContext {
      */
     @Nullable
     String partition();
+
+    /**
+     * Returns the table schema.
+     *
+     * @return the table schema
+     */
+    Schema schema();
 }

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/source/TieringSplitReader.java

@@ -301,7 +301,11 @@ private LakeWriter<WriteResult> getOrCreateLakeWriter(
         if (lakeWriter == null) {
             lakeWriter =
                     lakeTieringFactory.createLakeWriter(
-                            new TieringWriterInitContext(currentTablePath, bucket, partitionName));
+                            new TieringWriterInitContext(
+                                    currentTablePath,
+                                    bucket,
+                                    partitionName,
+                                    currentTable.getTableInfo().getSchema()));
             lakeWriters.put(bucket, lakeWriter);
         }
         return lakeWriter;

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/source/TieringWriterInitContext.java

@@ -18,6 +18,7 @@
 package com.alibaba.fluss.flink.tiering.source;
 
 import com.alibaba.fluss.lake.writer.WriterInitContext;
+import com.alibaba.fluss.metadata.Schema;
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TablePath;
 
@@ -28,17 +29,18 @@ public class TieringWriterInitContext implements WriterInitContext {
 
     private final TablePath tablePath;
     private final TableBucket tableBucket;
+    private final Schema schema;
     @Nullable private final String partition;
 
-    public TieringWriterInitContext(TablePath tablePath, TableBucket tableBucket) {
-        this(tablePath, tableBucket, null);
-    }
-
     public TieringWriterInitContext(
-            TablePath tablePath, TableBucket tableBucket, @Nullable String partition) {
+            TablePath tablePath,
+            TableBucket tableBucket,
+            @Nullable String partition,
+            Schema schema) {
         this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.partition = partition;
+        this.schema = schema;
     }
 
     @Override
@@ -56,4 +58,9 @@ public TableBucket tableBucket() {
     public String partition() {
         return partition;
     }
+
+    @Override
+    public Schema schema() {
+        return schema;
+    }
 }

fluss-lake/fluss-lake-paimon/src/test/java/com/alibaba/fluss/lake/paimon/tiering/PaimonTieringTest.java

@@ -658,6 +658,12 @@ public TableBucket tableBucket() {
                     public String partition() {
                         return partition;
                     }
+
+                    @Override
+                    public com.alibaba.fluss.metadata.Schema schema() {
+                        throw new UnsupportedOperationException(
+                                "The lake writer in Paimon currently uses paimonCatalog to determine the schema.");
+                    }
                 });
     }