diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
index 678a5a9c0dd7..9cf257deb058 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
@@ -37,8 +37,9 @@
 import org.apache.flink.table.catalog.CatalogPartition;
 import org.apache.flink.table.catalog.CatalogPartitionSpec;
 import org.apache.flink.table.catalog.CatalogTable;
-import org.apache.flink.table.catalog.CatalogTableImpl;
 import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
 import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
@@ -69,7 +70,6 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
-import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -77,6 +77,8 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}.
@@ -90,6 +92,8 @@
  */
 public class FlinkCatalog extends AbstractCatalog {
 
+  private static final Logger LOG = LoggerFactory.getLogger(FlinkCatalog.class);
+
   private final CatalogLoader catalogLoader;
   private final Catalog icebergCatalog;
   private final Namespace baseNamespace;
@@ -374,7 +378,7 @@ void createIcebergTable(ObjectPath tablePath, CatalogBaseTable table, boolean ig
       throws CatalogException, TableAlreadyExistException {
     validateFlinkTable(table);
 
-    Schema icebergSchema = FlinkSchemaUtil.convert(table.getSchema());
+    Schema icebergSchema = FlinkSchemaUtil.convert(((ResolvedCatalogTable) table).getResolvedSchema());
     PartitionSpec spec = toPartitionSpec(((CatalogTable) table).getPartitionKeys(), icebergSchema);
 
     ImmutableMap.Builder<String, String> properties = ImmutableMap.builder();
@@ -387,6 +391,11 @@ void createIcebergTable(ObjectPath tablePath, CatalogBaseTable table, boolean ig
       }
     }
 
+    // update properties from schema
+    Map<String, String> updatedProperties = FlinkSchemaUtil.generateTablePropertiesFromResolvedSchema(
+            ((ResolvedCatalogTable) table).getResolvedSchema());
+    properties.putAll(updatedProperties);
+
     try {
       icebergCatalog.createTable(
           toIdentifier(tablePath),
@@ -401,6 +410,19 @@ void createIcebergTable(ObjectPath tablePath, CatalogBaseTable table, boolean ig
     }
   }
 
+  private static void validateSchemaAndPartition(CatalogTable ct1, CatalogTable ct2) {
+    org.apache.flink.table.api.Schema ts1 = ct1.getUnresolvedSchema();
+    org.apache.flink.table.api.Schema ts2 = ct2.getUnresolvedSchema();
+
+    if (!Objects.equals(ts1, ts2)) {
+      throw new UnsupportedOperationException("Altering schema is not supported yet.");
+    }
+
+    if (!ct1.getPartitionKeys().equals(ct2.getPartitionKeys())) {
+      throw new UnsupportedOperationException("Altering partition keys is not supported yet.");
+    }
+  }
+
   private static void validateTableSchemaAndPartition(CatalogTable ct1, CatalogTable ct2) {
     TableSchema ts1 = ct1.getSchema();
     TableSchema ts2 = ct2.getSchema();
@@ -447,7 +469,7 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
 
     // For current Flink Catalog API, support for adding/removing/renaming columns cannot be done by comparing
     // CatalogTable instances, unless the Flink schema contains Iceberg column IDs.
-    validateTableSchemaAndPartition(table, (CatalogTable) newTable);
+    validateSchemaAndPartition(table, (CatalogTable) newTable);
 
     Map<String, String> oldProperties = table.getOptions();
     Map<String, String> setProperties = Maps.newHashMap();
@@ -481,22 +503,23 @@ public void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
       }
     });
 
+    Map<String, String> newTablePropertiesMap = Maps.newHashMap(oldProperties);
+    newTablePropertiesMap.putAll(setProperties);
+
+    CatalogTable catalogTable = toCatalogTable(icebergTable, newTablePropertiesMap);
+    ResolvedSchema resolvedSchema = FlinkSchemaUtil.convertToResolvedSchema(catalogTable);
+    Map<String, String> updatedProperties = FlinkSchemaUtil.generateTablePropertiesFromResolvedSchema(resolvedSchema);
+    setProperties.forEach((k, v) -> {
+      if (updatedProperties.containsKey(k)) {
+        setProperties.put(k, updatedProperties.get(k));
+      }
+    });
+
     commitChanges(icebergTable, setLocation, setSnapshotId, pickSnapshotId, setProperties);
   }
 
   private static void validateFlinkTable(CatalogBaseTable table) {
     Preconditions.checkArgument(table instanceof CatalogTable, "The Table should be a CatalogTable.");
-
-    TableSchema schema = table.getSchema();
-    schema.getTableColumns().forEach(column -> {
-      if (!FlinkCompatibilityUtil.isPhysicalColumn(column)) {
-        throw new UnsupportedOperationException("Creating table with computed columns is not supported yet.");
-      }
-    });
-
-    if (!schema.getWatermarkSpecs().isEmpty()) {
-      throw new UnsupportedOperationException("Creating table with watermark specs is not supported yet.");
-    }
   }
 
   private static PartitionSpec toPartitionSpec(List<String> partitionKeys, Schema icebergSchema) {
@@ -562,14 +585,33 @@ private static void commitChanges(Table table, String setLocation, String setSna
   }
 
   static CatalogTable toCatalogTable(Table table) {
-    TableSchema schema = FlinkSchemaUtil.toSchema(table.schema());
+
+    CatalogTable catalogTable = toCatalogTable(table, table.properties());
+    try {
+      FlinkSchemaUtil.convertToResolvedSchema(catalogTable);
+    } catch (RuntimeException e) {
+      LOG.warn("ignore watermark and computed columns!", e);
+      Map<String, String> properties = Maps.newHashMap();
+      table.properties().forEach((k, v) -> {
+        if (!k.startsWith(FlinkSchemaUtil.COMPUTED_COLUMNS_PREFIX) &&
+            !k.startsWith(FlinkSchemaUtil.WATERMARK_PREFIX)) {
+          properties.put(k, v);
+        }
+      });
+      catalogTable = toCatalogTable(table, properties);
+    }
+    return catalogTable;
+  }
+
+  static CatalogTable toCatalogTable(Table table, Map<String, String> properties) {
     List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema());
+    org.apache.flink.table.api.Schema schema = FlinkSchemaUtil.toSchema(table.schema(), properties);
 
     // NOTE: We can not create a IcebergCatalogTable extends CatalogTable, because Flink optimizer may use
     // CatalogTableImpl to copy a new catalog table.
     // Let's re-loading table from Iceberg catalog when creating source/sink operators.
     // Iceberg does not have Table comment, so pass a null (Default comment value in Flink).
-    return new CatalogTableImpl(schema, partitionKeys, table.properties(), null);
+    return CatalogTable.of(schema, null, partitionKeys, table.properties());
   }
 
   @Override
diff --git a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java
index 0827b21786c1..d614676125f3 100644
--- a/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java
+++ b/flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java
@@ -20,14 +20,25 @@
 package org.apache.iceberg.flink;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.internal.TableEnvironmentImpl;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.SchemaResolver;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.types.utils.TypeConversions;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
@@ -52,6 +63,14 @@
  */
 public class FlinkSchemaUtil {
 
+  public static final String FLINK_PREFIX = "flink.";
+
+  public static final String COMPUTED_COLUMNS = "computed-column.";
+  public static final String COMPUTED_COLUMNS_PREFIX = FLINK_PREFIX + COMPUTED_COLUMNS;
+
+  public static final String WATERMARK = "watermark.";
+  public static final String WATERMARK_PREFIX = FLINK_PREFIX + WATERMARK;
+
   private FlinkSchemaUtil() {
   }
 
@@ -59,28 +78,54 @@ private FlinkSchemaUtil() {
    * Convert the flink table schema to apache iceberg schema.
    */
   public static Schema convert(TableSchema schema) {
-    LogicalType schemaType = schema.toRowDataType().getLogicalType();
+    Schema iSchema = conv(schema.toPhysicalRowDataType().getLogicalType());
+    return freshIdentifierFieldIds(iSchema, schema);
+  }
+
+  /**
+   * Convert the flink table resolved schema to apache iceberg schema.
+   */
+  public static Schema convert(ResolvedSchema schema) {
+    Schema iSchema = conv(schema.toPhysicalRowDataType().getLogicalType());
+    return freshIdentifierFieldIds(iSchema, schema);
+  }
+
+  private static Schema conv(LogicalType schemaType) {
     Preconditions.checkArgument(schemaType instanceof RowType, "Schema logical type should be RowType.");
 
     RowType root = (RowType) schemaType;
     Type converted = root.accept(new FlinkTypeToType(root));
 
-    Schema iSchema = new Schema(converted.asStructType().fields());
-    return freshIdentifierFieldIds(iSchema, schema);
+    return new Schema(converted.asStructType().fields());
   }
 
   private static Schema freshIdentifierFieldIds(Schema iSchema, TableSchema schema) {
     // Locate the identifier field id list.
-    Set<Integer> identifierFieldIds = Sets.newHashSet();
+    List<String> primaryKeys = Lists.newArrayList();
     if (schema.getPrimaryKey().isPresent()) {
-      for (String column : schema.getPrimaryKey().get().getColumns()) {
+      primaryKeys.addAll(schema.getPrimaryKey().get().getColumns());
+    }
+    return freshIdentifier(iSchema, primaryKeys);
+  }
+
+  private static Schema freshIdentifierFieldIds(Schema iSchema, ResolvedSchema schema) {
+    List<String> primaryKeys = Lists.newArrayList();
+    if (schema.getPrimaryKey().isPresent()) {
+      primaryKeys.addAll(schema.getPrimaryKey().get().getColumns());
+    }
+    return freshIdentifier(iSchema, primaryKeys);
+  }
+
+  private static Schema freshIdentifier(Schema iSchema, List<String> keyColumns) {
+    Set<Integer> identifierFieldIds = Sets.newHashSet();
+    if (!keyColumns.isEmpty()) {
+      for (String column : keyColumns) {
         Types.NestedField field = iSchema.findField(column);
         Preconditions.checkNotNull(field,
             "Cannot find field ID for the primary key column %s in schema %s", column, iSchema);
         identifierFieldIds.add(field.fieldId());
       }
     }
-
     return new Schema(iSchema.schemaId(), iSchema.asStruct().fields(), identifierFieldIds);
   }
 
@@ -172,4 +217,110 @@ public static TableSchema toSchema(Schema schema) {
 
     return builder.build();
   }
+
+
+   /**
+   * Convert a {@link Schema} to a {@link Schema}.
+   *
+   * @param schema iceberg schema to convert.
+   * @return Flink Schema.
+   */
+  public static org.apache.flink.table.api.Schema toSchema(Schema schema, Map<String, String> properties) {
+
+    org.apache.flink.table.api.Schema.Builder builder = org.apache.flink.table.api.Schema.newBuilder();
+
+    // get watermark and computed columns
+    Map<String, String> watermarkMap = Maps.newHashMap();
+    Map<String, String> computedColumnsMap = Maps.newHashMap();
+    properties.keySet().stream()
+        .filter(k -> k.startsWith(FLINK_PREFIX) && properties.get(k) != null)
+        .forEach(k -> {
+          final String name = k.substring(k.lastIndexOf('.') + 1);
+          String expr = properties.get(k);
+          if (k.startsWith(WATERMARK_PREFIX)) {
+            watermarkMap.put(name, expr);
+          } else if (k.startsWith(COMPUTED_COLUMNS_PREFIX)) {
+            computedColumnsMap.put(name, expr);
+          }
+        });
+
+    // add physical columns.
+    for (RowType.RowField field : convert(schema).getFields()) {
+      builder.column(field.getName(), TypeConversions.fromLogicalToDataType(field.getType()));
+    }
+
+    // add computed columns.
+    computedColumnsMap.forEach(builder::columnByExpression);
+
+    // add watermarks.
+    watermarkMap.forEach(builder::watermark);
+
+    // add primary key.
+    List<String> primaryKey = getPrimaryKeyFromSchema(schema);
+    if (!primaryKey.isEmpty()) {
+      builder.primaryKey(primaryKey.toArray(new String[0]));
+    }
+
+    return builder.build();
+  }
+
+  /**
+   * Convert a {@link CatalogTable} to a {@link ResolvedSchema}.
+   *
+   * @param table flink unresolved schema to convert.
+   * @return Flink ResolvedSchema.
+   */
+  public static ResolvedSchema convertToResolvedSchema(CatalogTable table) {
+    Configuration configuration = ExecutionEnvironment.getExecutionEnvironment().getConfiguration();
+    TableEnvironment tableEnvironment = TableEnvironment.create(configuration);
+    CatalogManager catalogManager = ((TableEnvironmentImpl) tableEnvironment).getCatalogManager();
+    SchemaResolver schemaResolver = catalogManager.getSchemaResolver();
+    return table.getUnresolvedSchema().resolve(schemaResolver);
+  }
+
+  /**
+   * Generate table properties for watermark and computed columns from flink resolved schema.
+   *
+   * @param schema flink resolved schema.
+   * @return Table properties map.
+   */
+  public static Map<String, String> generateTablePropertiesFromResolvedSchema(ResolvedSchema schema) {
+    Map<String, String> properties = Maps.newHashMap();
+
+    // save watermark
+    schema.getWatermarkSpecs().forEach(column -> {
+      String name = column.getRowtimeAttribute();
+      properties.put(
+          FlinkSchemaUtil.WATERMARK_PREFIX + name,
+          column.getWatermarkExpression().asSerializableString());
+    });
+
+    // save computed columns
+    schema.getColumns().stream()
+        .filter(column -> column instanceof Column.ComputedColumn)
+        .forEach(tableColumn -> {
+          Column.ComputedColumn column = (Column.ComputedColumn) tableColumn;
+          String name = column.getName();
+          properties.put(
+              FlinkSchemaUtil.COMPUTED_COLUMNS_PREFIX + name,
+              column.getExpression().asSerializableString());
+        });
+
+    return properties;
+  }
+
+  public static List<String> getPrimaryKeyFromSchema(Schema schema) {
+    Set<Integer> identifierFieldIds = schema.identifierFieldIds();
+    if (!identifierFieldIds.isEmpty()) {
+      List<String> columns = Lists.newArrayListWithExpectedSize(identifierFieldIds.size());
+      for (Integer identifierFieldId : identifierFieldIds) {
+        String columnName = schema.findColumnName(identifierFieldId);
+        Preconditions.checkNotNull(columnName,
+                "Cannot find field with id %s in schema %s", identifierFieldId, schema);
+        columns.add(columnName);
+      }
+      return columns;
+    }
+    return Lists.newArrayList();
+  }
 }
diff --git a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
index 7f8eac9d2c8b..54bcdba420c4 100644
--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/FlinkTestBase.java
@@ -23,6 +23,7 @@
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.config.TableConfigOptions;
 import org.apache.flink.test.util.MiniClusterWithClientResource;
 import org.apache.flink.test.util.TestBaseUtils;
 import org.apache.flink.types.Row;
@@ -78,7 +79,9 @@ protected TableEnvironment getTableEnv() {
               .build();
 
           TableEnvironment env = TableEnvironment.create(settings);
-          env.getConfig().getConfiguration().set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false);
+          env.getConfig().getConfiguration()
+              .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, false)
+              .set(TableConfigOptions.LOCAL_TIME_ZONE, "UTC");
           tEnv = env;
         }
       }
diff --git a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
index 37b45a69777e..a259059d23d0 100644
--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java
@@ -21,18 +21,20 @@
 
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Schema.UnresolvedPrimaryKey;
+import org.apache.flink.table.api.TableException;
 import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.api.constraints.UniqueConstraint;
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.catalog.ObjectPath;
 import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.types.Row;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.ContentFile;
@@ -50,8 +52,8 @@
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Types;
@@ -61,10 +63,13 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import static org.apache.iceberg.flink.FlinkSchemaUtil.COMPUTED_COLUMNS_PREFIX;
+import static org.apache.iceberg.flink.FlinkSchemaUtil.WATERMARK_PREFIX;
+
 public class TestFlinkCatalogTable extends FlinkCatalogTestBase {
 
-  public TestFlinkCatalogTable(String catalogName, Namespace baseNamepace) {
-    super(catalogName, baseNamepace);
+  public TestFlinkCatalogTable(String catalogName, Namespace baseNamespace) {
+    super(catalogName, baseNamespace);
   }
 
   @Override
@@ -109,7 +114,7 @@ public void testRenameTable() {
         "Table `tl` was not found.",
         () -> getTableEnv().from("tl")
     );
-    Schema actualSchema = FlinkSchemaUtil.convert(getTableEnv().from("tl2").getSchema());
+    Schema actualSchema = FlinkSchemaUtil.convert(getTableEnv().from("tl2").getResolvedSchema());
     Assert.assertEquals(tableSchema.asStruct(), actualSchema.asStruct());
   }
 
@@ -124,7 +129,9 @@ public void testCreateTable() throws TableNotExistException {
     Assert.assertEquals(Maps.newHashMap(), table.properties());
 
     CatalogTable catalogTable = catalogTable("tl");
-    Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+    Assert.assertEquals(
+        org.apache.flink.table.api.Schema.newBuilder().column("id", DataTypes.BIGINT()).build(),
+        catalogTable.getUnresolvedSchema());
     Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
   }
 
@@ -138,10 +145,10 @@ public void testCreateTableWithPrimaryKey() throws Exception {
         table.schema().identifierFieldIds());
 
     CatalogTable catalogTable = catalogTable("tl");
-    Optional<UniqueConstraint> uniqueConstraintOptional = catalogTable.getSchema().getPrimaryKey();
-    Assert.assertTrue("Should have the expected unique constraint", uniqueConstraintOptional.isPresent());
+    Optional<UnresolvedPrimaryKey> primaryKey = catalogTable.getUnresolvedSchema().getPrimaryKey();
+    Assert.assertTrue("Should have the expected unique constraint", primaryKey.isPresent());
     Assert.assertEquals("Should have the expected columns",
-        ImmutableList.of("key"), uniqueConstraintOptional.get().getColumns());
+        ImmutableList.of("key"), primaryKey.get().getColumnNames());
   }
 
   @Test
@@ -156,10 +163,10 @@ public void testCreateTableWithMultiColumnsInPrimaryKey() throws Exception {
         table.schema().identifierFieldIds());
 
     CatalogTable catalogTable = catalogTable("tl");
-    Optional<UniqueConstraint> uniqueConstraintOptional = catalogTable.getSchema().getPrimaryKey();
-    Assert.assertTrue("Should have the expected unique constraint", uniqueConstraintOptional.isPresent());
+    Optional<UnresolvedPrimaryKey> primaryKey = catalogTable.getUnresolvedSchema().getPrimaryKey();
+    Assert.assertTrue("Should have the expected unique constraint", primaryKey.isPresent());
     Assert.assertEquals("Should have the expected columns",
-        ImmutableSet.of("data", "id"), ImmutableSet.copyOf(uniqueConstraintOptional.get().getColumns()));
+        ImmutableList.of("id", "data"), primaryKey.get().getColumnNames());
   }
 
   @Test
@@ -201,7 +208,9 @@ public void testCreateTableLike() throws TableNotExistException {
     Assert.assertEquals(Maps.newHashMap(), table.properties());
 
     CatalogTable catalogTable = catalogTable("tl2");
-    Assert.assertEquals(TableSchema.builder().field("id", DataTypes.BIGINT()).build(), catalogTable.getSchema());
+    Assert.assertEquals(
+        org.apache.flink.table.api.Schema.newBuilder().column("id", DataTypes.BIGINT()).build(),
+        catalogTable.getUnresolvedSchema());
     Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
   }
 
@@ -234,8 +243,11 @@ public void testCreatePartitionTable() throws TableNotExistException {
 
     CatalogTable catalogTable = catalogTable("tl");
     Assert.assertEquals(
-        TableSchema.builder().field("id", DataTypes.BIGINT()).field("dt", DataTypes.STRING()).build(),
-        catalogTable.getSchema());
+        org.apache.flink.table.api.Schema.newBuilder()
+            .column("id", DataTypes.BIGINT())
+            .column("dt", DataTypes.STRING()).build(),
+        catalogTable.getUnresolvedSchema()
+    );
     Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
     Assert.assertEquals(Collections.singletonList("dt"), catalogTable.getPartitionKeys());
   }
@@ -287,8 +299,9 @@ public void testLoadTransformPartitionTable() throws TableNotExistException {
 
     CatalogTable catalogTable = catalogTable("tl");
     Assert.assertEquals(
-        TableSchema.builder().field("id", DataTypes.BIGINT()).build(),
-        catalogTable.getSchema());
+        org.apache.flink.table.api.Schema.newBuilder().column("id", DataTypes.BIGINT()).build(),
+        catalogTable.getUnresolvedSchema()
+    );
     Assert.assertEquals(Maps.newHashMap(), catalogTable.getOptions());
     Assert.assertEquals(Collections.emptyList(), catalogTable.getPartitionKeys());
   }
@@ -420,4 +433,166 @@ private CatalogTable catalogTable(String name) throws TableNotExistException {
     return (CatalogTable) getTableEnv().getCatalog(getTableEnv().getCurrentCatalog()).get()
         .getTable(new ObjectPath(DATABASE, name));
   }
+
+  private Map<String, String> createTableWithoutPrimaryKey() {
+    sql("create table tl (\n" +
+        "id int, \n" +
+        "id2 as id * 2, \n" +
+        "s varchar(10), \n" +
+        "f1 as TO_TIMESTAMP(FROM_UNIXTIME(id*3)), \n" +
+        "t1 timestamp(6), \n" +
+        "t2 as cast(t1 as timestamp(3)), \n" +
+        "watermark for t2 as t2 - INTERVAL '5' SECOND )");
+
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put("flink.computed-column.id2", "`id` * 2");
+    properties.put("flink.computed-column.f1", "TO_TIMESTAMP(FROM_UNIXTIME(`id` * 3))");
+    properties.put("flink.computed-column.t2", "CAST(`t1` AS TIMESTAMP(3))");
+    properties.put("flink.watermark.t2", "`t2` - INTERVAL '5' SECOND");
+    Assert.assertEquals(properties, table("tl").properties());
+
+    return properties;
+  }
+
+  private Map<String, String> createTableWithPrimaryKey() {
+    sql("create table tl (\n" +
+        "id int, \n" +
+        "id2 as id * 2, \n" +
+        "s varchar(10), \n" +
+        "f1 as TO_TIMESTAMP(FROM_UNIXTIME(id*3)), \n" +
+        "t1 timestamp(6), \n" +
+        "primary key (id, s) not enforced, \n" +
+        "t2 as cast(t1 as timestamp(2)), \n" +
+        "watermark for t2 as t2 - INTERVAL '5' SECOND )");
+
+    Map<String, String> properties = Maps.newHashMap();
+    properties.put("flink.computed-column.id2", "`id` * 2");
+    properties.put("flink.computed-column.f1", "TO_TIMESTAMP(FROM_UNIXTIME(`id` * 3))");
+    properties.put("flink.computed-column.t2", "CAST(`t1` AS TIMESTAMP(2))");
+    properties.put("flink.watermark.t2", "`t2` - INTERVAL '5' SECOND");
+    Assert.assertEquals(properties, table("tl").properties());
+
+    return properties;
+  }
+
+  @Test
+  public void testComputedColumnsWithoutPrimaryKey() {
+    Map<String, String> properties = createTableWithoutPrimaryKey();
+    testComputedColumns(properties);
+  }
+
+  @Test
+  public void testComputedColumnsWithPrimaryKey() {
+    Map<String, String> properties = createTableWithPrimaryKey();
+    testComputedColumns(properties);
+  }
+
+  private void testComputedColumns(Map<String, String> properties) {
+    String id2Key = COMPUTED_COLUMNS_PREFIX + "id2";
+
+    // reset id2, success
+    sql("ALTER TABLE tl RESET ('" + id2Key + "')");
+    properties.remove(id2Key);
+    Assert.assertEquals(properties, table("tl").properties());
+
+    // add id2, success
+    sql("ALTER TABLE tl SET ('" + id2Key + "'= 'id*3')");
+    properties.put(id2Key, "`id` * 3");
+    Assert.assertEquals(properties, table("tl").properties());
+
+    // update id2, success
+    sql("ALTER TABLE tl SET ('" + id2Key + "'='id*4')");
+    properties.put(id2Key, "`id` * 4");
+    Assert.assertEquals(properties, table("tl").properties());
+
+    // update id2, depend on column which not exist, failed
+    AssertHelpers.assertThrows("should throw TableException.",
+        TableException.class,
+        () -> sql("ALTER TABLE tl SET ('" + id2Key + "'='ab*4')"));
+
+    // update id2, error expr, failed
+    AssertHelpers.assertThrows("should throw TableException.",
+        TableException.class,
+        () -> sql("ALTER TABLE tl SET ('" + id2Key + "'='error*4')"));
+  }
+
+  @Test
+  public void testWatermarkWithoutPrimaryKey() {
+    Map<String, String> properties = createTableWithoutPrimaryKey();
+    testWatermark(properties);
+  }
+
+  @Test
+  public void testWatermarkWithPrimaryKey() {
+    Map<String, String> properties = createTableWithPrimaryKey();
+    testWatermark(properties);
+  }
+
+  private void testWatermark(Map<String, String> properties) {
+    String t2WaterMark = WATERMARK_PREFIX + "t2";
+
+    // add, error because the watermark already exists
+    AssertHelpers.assertThrows("should throw TableException.",
+        TableException.class,
+        () -> sql("ALTER TABLE tl SET ('flink.watermark.t3'='t2 - INTERVAL ''5'' SECOND')"));
+
+    // reset, success
+    sql("ALTER TABLE tl RESET ('" + t2WaterMark + "')");
+    properties.remove(t2WaterMark);
+    Assert.assertEquals(properties, table("tl").properties());
+
+    // add, success
+    sql("ALTER TABLE tl SET ('" + t2WaterMark + "'='t2 - INTERVAL ''15'' SECOND')");
+    properties.put(t2WaterMark, "`t2` - INTERVAL '15' SECOND");
+
+    // update, success
+    sql("ALTER TABLE tl SET ('" + t2WaterMark + "'='t2 - INTERVAL ''25'' SECOND')");
+    properties.put(t2WaterMark, "`t2` - INTERVAL '25' SECOND");
+    Assert.assertEquals(properties, table("tl").properties());
+
+    // reset computed column t2, error because watermark t2 depend on computed column t2
+    AssertHelpers.assertThrows("should throw TableException.",
+        TableException.class,
+        () -> sql("ALTER TABLE tl RESET ('flink.computed-column.t2')"));
+  }
+
+  @Test
+  public void testTableDataWithWatermarkAndComputedColumns() {
+    createTableWithoutPrimaryKey();
+
+    sql("INSERT INTO tl VALUES (1, 'abc', TO_TIMESTAMP(FROM_UNIXTIME(24)))");
+    List<String> expectResult = Lists.newArrayList();
+    expectResult.add("1");
+    expectResult.add("2");
+    expectResult.add("abc");
+    expectResult.add("1970-01-01T00:00:03");
+    expectResult.add("1970-01-01T00:00:24");
+    expectResult.add("1970-01-01T00:00:24");
+
+    List<String> result = Lists.newArrayList();
+    Row row = sql("SELECT id, id2, s, f1, t1, t2 FROM tl").get(0);
+    for (int i = 0; i < 6; i++) {
+      Object field = row.getField(i);
+      assert field != null;
+      result.add(field.toString());
+    }
+    Assert.assertEquals(expectResult, result);
+
+    // drop the physical column that the computed column depends on,
+    // the table will return physical columns only
+    table("tl").updateSchema().deleteColumn("id").commit();
+
+    List<String> expectResult2 = Lists.newArrayList();
+    expectResult2.add("abc");
+    expectResult2.add("1970-01-01T00:00:24");
+
+    List<String> result2 = Lists.newArrayList();
+    Row row2 = sql("SELECT * FROM tl").get(0);
+    for (int i = 0; i < 2; i++) {
+      Object field = row2.getField(i);
+      assert field != null;
+      result2.add(field.toString());
+    }
+    Assert.assertEquals(expectResult2, result2);
+  }
 }
diff --git a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java
index 01f8524464e0..75280b49bb03 100644
--- a/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java
+++ b/flink/v1.14/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java
@@ -35,6 +35,7 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
@@ -306,10 +307,10 @@ public void testConvertFlinkSchemaWithPrimaryKeys() {
         Sets.newHashSet(1, 2)
     );
 
-    TableSchema tableSchema = FlinkSchemaUtil.toSchema(icebergSchema);
+    org.apache.flink.table.api.Schema tableSchema = FlinkSchemaUtil.toSchema(icebergSchema, Maps.newHashMap());
     Assert.assertTrue(tableSchema.getPrimaryKey().isPresent());
     Assert.assertEquals(ImmutableSet.of("int", "string"),
-        ImmutableSet.copyOf(tableSchema.getPrimaryKey().get().getColumns()));
+        ImmutableSet.copyOf(tableSchema.getPrimaryKey().get().getColumnNames()));
   }
 
   @Test