diff --git a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUtil.java b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUtil.java
index b5e8f6662cc69..fb56644872728 100644
--- a/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUtil.java
+++ b/presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergUtil.java
@@ -30,8 +30,10 @@
 import com.facebook.presto.hive.HdfsContext;
 import com.facebook.presto.hive.HdfsEnvironment;
 import com.facebook.presto.hive.HiveColumnConverterProvider;
+import com.facebook.presto.hive.HiveCompressionCodec;
 import com.facebook.presto.hive.HivePartition;
 import com.facebook.presto.hive.HivePartitionKey;
+import com.facebook.presto.hive.HiveStorageFormat;
 import com.facebook.presto.hive.HiveType;
 import com.facebook.presto.hive.PartitionNameWithVersion;
 import com.facebook.presto.hive.metastore.Column;
@@ -1163,12 +1165,19 @@ public static Map<String, String> populateTableProperties(IcebergAbstractMetadat
         Integer commitRetries = tableProperties.getCommitRetries(session, tableMetadata.getProperties());
         propertiesBuilder.put(DEFAULT_FILE_FORMAT, fileFormat.toString());
         propertiesBuilder.put(COMMIT_NUM_RETRIES, String.valueOf(commitRetries));
+        HiveCompressionCodec compressionCodec = getCompressionCodec(session);
         switch (fileFormat) {
             case PARQUET:
-                propertiesBuilder.put(PARQUET_COMPRESSION, getCompressionCodec(session).getParquetCompressionCodec().get().toString());
+                if (!compressionCodec.isSupportedStorageFormat(HiveStorageFormat.PARQUET)) {
+                    throw new PrestoException(NOT_SUPPORTED, format("Compression codec %s is not supported for Parquet format", compressionCodec));
+                }
+                propertiesBuilder.put(PARQUET_COMPRESSION, compressionCodec.getParquetCompressionCodec().get().toString());
                 break;
             case ORC:
-                propertiesBuilder.put(ORC_COMPRESSION, getCompressionCodec(session).getOrcCompressionKind().name());
+                if (!compressionCodec.isSupportedStorageFormat(HiveStorageFormat.ORC)) {
+                    throw new PrestoException(NOT_SUPPORTED, format("Compression codec %s is not supported for ORC format", compressionCodec));
+                }
+                propertiesBuilder.put(ORC_COMPRESSION, compressionCodec.getOrcCompressionKind().name());
                 break;
         }
         if (tableMetadata.getComment().isPresent()) {
diff --git a/presto-iceberg/src/test/java/com/facebook/presto/iceberg/IcebergDistributedSmokeTestBase.java b/presto-iceberg/src/test/java/com/facebook/presto/iceberg/IcebergDistributedSmokeTestBase.java
index 052e65618fbb3..97ef9bb9ff1f3 100644
--- a/presto-iceberg/src/test/java/com/facebook/presto/iceberg/IcebergDistributedSmokeTestBase.java
+++ b/presto-iceberg/src/test/java/com/facebook/presto/iceberg/IcebergDistributedSmokeTestBase.java
@@ -1106,6 +1106,41 @@ protected void unregisterTable(String schemaName, String newTableName)
         assertUpdate("CALL system.unregister_table('" + schemaName + "', '" + newTableName + "')");
     }
 
+    @DataProvider
+    public Object[][] compressionCodecTestData()
+    {
+        return new Object[][] {
+                // codec, format, shouldSucceed, expectedErrorMessage
+                {"ZSTD", "PARQUET", false, "Compression codec ZSTD is not supported for Parquet format"},
+                {"LZ4", "PARQUET", false, "Compression codec LZ4 is not supported for Parquet format"},
+                {"LZ4", "ORC", false, "Compression codec LZ4 is not supported for ORC format"},
+                {"ZSTD", "ORC", true, null},
+                {"SNAPPY", "ORC", true, null},
+                {"SNAPPY", "PARQUET", true, null},
+                {"GZIP", "PARQUET", true, null},
+                {"NONE", "PARQUET", true, null},
+        };
+    }
+
+    @Test(dataProvider = "compressionCodecTestData")
+    public void testCompressionCodecValidation(String codec, String format, boolean shouldSucceed, String expectedErrorMessage)
+    {
+        Session session = Session.builder(getSession())
+                .setCatalogSessionProperty("iceberg", "compression_codec", codec)
+                .build();
+
+        String tableName = "test_compression_" + codec.toLowerCase() + "_" + format.toLowerCase();
+        String createTableSql = "CREATE TABLE " + tableName + " (x int) WITH (format = '" + format + "')";
+
+        if (shouldSucceed) {
+            assertUpdate(session, createTableSql);
+            dropTable(session, tableName);
+        }
+        else {
+            assertQueryFails(session, createTableSql, expectedErrorMessage);
+        }
+    }
+
     @Test
     public void testCreateNestedPartitionedTable()
     {
diff --git a/presto-iceberg/src/test/java/com/facebook/presto/iceberg/TestIcebergUtil.java b/presto-iceberg/src/test/java/com/facebook/presto/iceberg/TestIcebergUtil.java
index 47c06c962a9e4..f8ed4bc75be9e 100644
--- a/presto-iceberg/src/test/java/com/facebook/presto/iceberg/TestIcebergUtil.java
+++ b/presto-iceberg/src/test/java/com/facebook/presto/iceberg/TestIcebergUtil.java
@@ -14,6 +14,9 @@
 package com.facebook.presto.iceberg;
 
 import com.facebook.presto.common.type.DecimalType;
+import com.facebook.presto.hive.HiveCompressionCodec;
+import com.facebook.presto.hive.HiveStorageFormat;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 import java.math.BigDecimal;
@@ -377,4 +380,41 @@ public void testGetTargetSplitSize()
         assertEquals(1024, getTargetSplitSize(1024, 512).toBytes());
         assertEquals(512, getTargetSplitSize(0, 512).toBytes());
     }
+
+    @DataProvider
+    public Object[][] compressionCodecMatrix()
+    {
+        return new Object[][] {
+                // format, codec, expectedSupport
+                {HiveStorageFormat.PARQUET, HiveCompressionCodec.NONE, true},
+                {HiveStorageFormat.PARQUET, HiveCompressionCodec.SNAPPY, true},
+                {HiveStorageFormat.PARQUET, HiveCompressionCodec.GZIP, true},
+                {HiveStorageFormat.PARQUET, HiveCompressionCodec.LZ4, false},
+                {HiveStorageFormat.PARQUET, HiveCompressionCodec.ZSTD, false},
+                {HiveStorageFormat.ORC, HiveCompressionCodec.NONE, true},
+                {HiveStorageFormat.ORC, HiveCompressionCodec.SNAPPY, true},
+                {HiveStorageFormat.ORC, HiveCompressionCodec.GZIP, true},
+                {HiveStorageFormat.ORC, HiveCompressionCodec.ZSTD, true},
+                {HiveStorageFormat.ORC, HiveCompressionCodec.LZ4, false},
+        };
+    }
+
+    @Test(dataProvider = "compressionCodecMatrix")
+    public void testCompressionCodecSupport(HiveStorageFormat format, HiveCompressionCodec codec, boolean expectedSupport)
+    {
+        assertThat(codec.isSupportedStorageFormat(format))
+                .as("Codec %s support for %s format", codec, format)
+                .isEqualTo(expectedSupport);
+    }
+
+    @Test
+    public void testParquetCompressionCodecAvailability()
+    {
+        assertThat(HiveCompressionCodec.NONE.getParquetCompressionCodec().isPresent()).isTrue();
+        assertThat(HiveCompressionCodec.SNAPPY.getParquetCompressionCodec().isPresent()).isTrue();
+        assertThat(HiveCompressionCodec.GZIP.getParquetCompressionCodec().isPresent()).isTrue();
+
+        assertThat(HiveCompressionCodec.LZ4.getParquetCompressionCodec().isPresent()).isFalse();
+        assertThat(HiveCompressionCodec.ZSTD.getParquetCompressionCodec().isPresent()).isFalse();
+    }
 }