Address amoghs feedback

Author: sfc-gh-prsingh

api/src/main/java/org/apache/iceberg/expressions/Literals.java

@@ -30,6 +30,7 @@
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
 import java.util.Comparator;
+import java.util.Locale;
 import java.util.Objects;
 import java.util.UUID;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -587,6 +588,32 @@ public <T> Literal<T> to(Type type) {
           BigDecimal decimal = new BigDecimal(value().toString());
           return (Literal<T>) new DecimalLiteral(decimal);
 
+        case FIXED:
+          try {
+            ByteBuffer b =
+                ByteBuffer.wrap(
+                    BaseEncoding.base16().decode(value().toString().toUpperCase(Locale.ROOT)));
+            Types.FixedType fixed = (Types.FixedType) type;
+            if (b.remaining() == fixed.length()) {
+              return (Literal<T>) new FixedLiteral(b);
+            }
+            return null;
+          } catch (IllegalArgumentException e) {
+            // Invalid hex string
+            return null;
+          }
+
+        case BINARY:
+          try {
+            return (Literal<T>)
+                new BinaryLiteral(
+                    ByteBuffer.wrap(
+                        BaseEncoding.base16().decode(value().toString().toUpperCase(Locale.ROOT))));
+          } catch (IllegalArgumentException e) {
+            // Invalid hex string
+            return null;
+          }
+
         default:
           return null;
       }

spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRemoteScanPlanning.java

@@ -25,8 +25,6 @@
 import org.apache.iceberg.rest.RESTCatalogProperties;
 import org.apache.iceberg.spark.SparkCatalogConfig;
 import org.apache.iceberg.spark.sql.TestSelect;
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.api.TestTemplate;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 @ExtendWith(ParameterizedTestExtension.class)
@@ -48,11 +46,4 @@ protected static Object[][] parameters() {
       }
     };
   }
-
-  @TestTemplate
-  @Disabled(
-      "binary filter that is used by Spark is not working because ExpressionParser.fromJSON doesn't have the Schema to properly parse the filter expression")
-  public void testBinaryInFilter() {
-    super.testBinaryInFilter();
-  }
 }

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java

@@ -32,7 +32,10 @@
 import org.apache.iceberg.Parameter;
 import org.apache.iceberg.ParameterizedTestExtension;
 import org.apache.iceberg.Parameters;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.events.Listeners;
 import org.apache.iceberg.events.ScanEvent;
 import org.apache.iceberg.exceptions.ValidationException;
@@ -43,6 +46,7 @@
 import org.apache.iceberg.spark.Spark3Util;
 import org.apache.iceberg.spark.SparkCatalogConfig;
 import org.apache.iceberg.spark.SparkReadOptions;
+import org.apache.iceberg.types.Types;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.junit.jupiter.api.AfterEach;
@@ -637,6 +641,29 @@ public void testBinaryInFilter() {
         sql("SELECT id, binary FROM %s where binary > X'11'", binaryTableName));
   }
 
+  @TestTemplate
+  public void testFixedInFilter() {
+    // Create table programmatically with fixed type since Spark SQL DDL doesn't support it
+    Schema schema =
+        new Schema(
+            Types.NestedField.required(1, "id", Types.LongType.get()),
+            Types.NestedField.required(2, "fixed", Types.FixedType.ofLength(2)));
+
+    TableIdentifier fixedTableIdent = TableIdentifier.of(tableIdent.namespace(), "fixed_table");
+    validationCatalog.createTable(fixedTableIdent, schema, PartitionSpec.unpartitioned());
+
+    String fixedTableName = tableName("fixed_table");
+    sql("INSERT INTO %s VALUES (1, X'0000'), (2, X'1111'), (3, X'0011')", fixedTableName);
+    List<Object[]> expected = ImmutableList.of(row(2L, new byte[] {0x11, 0x11}));
+
+    assertEquals(
+        "Should return all expected rows",
+        expected,
+        sql("SELECT id, fixed FROM %s WHERE fixed > X'0011'", fixedTableName));
+
+    sql("DROP TABLE IF EXISTS %s", fixedTableName);
+  }
+
   @TestTemplate
   public void testComplexTypeFilter() {
     String complexTypeTableName = tableName("complex_table");