[lake] pass table custom table property to lake writer (#1521)

Author: xx789633

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

@@ -24,6 +24,8 @@
 
 import javax.annotation.Nullable;
 
+import java.util.Map;
+
 /**
  * The WriterInitContext interface provides the context needed to create a LakeWriter. It includes
  * methods to obtain the table path, table bucket, and an optional partition.
@@ -61,4 +63,11 @@ public interface WriterInitContext {
      * @return the table schema
      */
     Schema schema();
+
+    /**
+     * Returns the table custom properties.
+     *
+     * @return the table custom properties
+     */
+    Map<String, String> customProperties();
 }

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

@@ -309,7 +309,8 @@ private LakeWriter<WriteResult> getOrCreateLakeWriter(
                                     currentTablePath,
                                     bucket,
                                     partitionName,
-                                    currentTable.getTableInfo().getSchema()));
+                                    currentTable.getTableInfo().getSchema(),
+                                    currentTable.getTableInfo().getCustomProperties().toMap()));
             lakeWriters.put(bucket, lakeWriter);
         }
         return lakeWriter;

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

@@ -24,23 +24,28 @@
 
 import javax.annotation.Nullable;
 
+import java.util.Map;
+
 /** The implementation of {@link WriterInitContext}. */
 public class TieringWriterInitContext implements WriterInitContext {
 
     private final TablePath tablePath;
     private final TableBucket tableBucket;
     private final Schema schema;
     @Nullable private final String partition;
+    private final Map<String, String> customProperties;
 
     public TieringWriterInitContext(
             TablePath tablePath,
             TableBucket tableBucket,
             @Nullable String partition,
-            Schema schema) {
+            Schema schema,
+            Map<String, String> customProperties) {
         this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.partition = partition;
         this.schema = schema;
+        this.customProperties = customProperties;
     }
 
     @Override
@@ -63,4 +68,9 @@ public String partition() {
     public Schema schema() {
         return schema;
     }
+
+    @Override
+    public Map<String, String> customProperties() {
+        return customProperties;
+    }
 }

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

@@ -716,6 +716,12 @@ public String partition() {
                         return partition;
                     }
 
+                    @Override
+                    public Map<String, String> customProperties() {
+                        // don't care about table custom properties for Paimon lake writer
+                        return new HashMap<>();
+                    }
+
                     @Override
                     public com.alibaba.fluss.metadata.Schema schema() {
                         throw new UnsupportedOperationException(