Reject varbinary partition columns in Delta Lake connector

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

@@ -1748,6 +1748,13 @@ private static void checkPartitionColumns(List<ColumnMetadata> columns, List<Str
             throw new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Using array, map or row type on partitioned columns is unsupported");
         }
 
+        // The Delta Lake protocol stores binary partition values as URI-encoded bytes in directory names,
+        // which the existing write path does not implement; reject at schema validation rather than failing later at INSERT time.
+        if (columns.stream().filter(column -> partitionColumnNames.contains(column.getName()))
+                .anyMatch(column -> column.getType().equals(VARBINARY))) {
+            throw new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Using varbinary type on partitioned columns is unsupported");
+        }
+
         if (!invalidPartitionNames.isEmpty()) {
             throw new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Table property 'partitioned_by' contained column names which do not exist: " + invalidPartitionNames);
         }

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeBasic.java

@@ -548,7 +548,7 @@ void testPartitionValuesParsedCheckpoint()
                     ImmutableList.of("0.12", "3.45"),
                     ImmutableList.of(decimal("0.12", createDecimalType(3, 2)), decimal("3.45", createDecimalType(3, 2))));
             testPartitionValuesParsedCheckpoint(mode, "varchar", ImmutableList.of("'alice'", "'bob'"), ImmutableList.of("alice", "bob"));
-            // TODO https://github.com/trinodb/trino/issues/24155 Cannot insert varbinary values into partitioned columns
+            // varbinary partition columns are rejected at schema validation; see testVarbinaryPartitionColumnRejected.
             testPartitionValuesParsedCheckpoint(
                     mode,
                     "date",
@@ -628,6 +628,18 @@ private void testPartitionValuesParsedCheckpoint(ColumnMappingMode columnMapping
         }
     }
 
+    @Test // https://github.com/trinodb/trino/issues/24155
+    void testVarbinaryPartitionColumnRejected()
+    {
+        String tableName = "test_varbinary_partition_" + randomNameSuffix();
+        assertQueryFails(
+                "CREATE TABLE " + tableName + "(x INT, part VARBINARY) WITH (partitioned_by = ARRAY['part'])",
+                "Using varbinary type on partitioned columns is unsupported");
+        assertQueryFails(
+                "CREATE TABLE " + tableName + " WITH (partitioned_by = ARRAY['part']) AS SELECT 1 x, X'01' part",
+                "Using varbinary type on partitioned columns is unsupported");
+    }
+
     /**
      * @see deltalake.column_mapping_mode_id
      * @see deltalake.column_mapping_mode_name

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConnectorTest.java

@@ -458,18 +458,9 @@ public void testCreateTableWithUnsupportedPartitionType()
         assertQueryFails(
                 "CREATE TABLE " + tableName + "(a INT, part ROW(field INT)) WITH (partitioned_by = ARRAY['part'])",
                 "Using array, map or row type on partitioned columns is unsupported");
-    }
-
-    @Test
-    public void testInsertIntoUnsupportedVarbinaryPartitionType()
-    {
-        // TODO https://github.com/trinodb/trino/issues/24155 Cannot insert varbinary values into partitioned columns
-        // Update TestDeltaLakeBasic.testPartitionValuesParsedCheckpoint() when fixing this issue
-        try (TestTable table = newTrinoTable(
-                "test_varbinary_partition",
-                "(x int, part varbinary) WITH (partitioned_by = ARRAY['part'])")) {
-            assertQueryFails("INSERT INTO " + table.getName() + " VALUES (1, X'01')", "Unsupported type for partition: varbinary");
-        }
+        assertQueryFails(
+                "CREATE TABLE " + tableName + "(a INT, part VARBINARY) WITH (partitioned_by = ARRAY['part'])",
+                "Using varbinary type on partitioned columns is unsupported");
     }
 
     @Test
@@ -485,6 +476,9 @@ public void testCreateTableAsSelectWithUnsupportedPartitionType()
         assertQueryFails(
                 "CREATE TABLE " + tableName + " WITH (partitioned_by = ARRAY['part']) AS SELECT 1 a, row(1) part",
                 "Using array, map or row type on partitioned columns is unsupported");
+        assertQueryFails(
+                "CREATE TABLE " + tableName + " WITH (partitioned_by = ARRAY['part']) AS SELECT 1 a, X'01' part",
+                "Using varbinary type on partitioned columns is unsupported");
     }
 
     @Test