Address review feeback

Author: sfc-gh-prsingh

core/src/main/java/org/apache/iceberg/SingleValueParser.java

@@ -119,14 +119,41 @@ public static Object fromJson(Type type, JsonNode defaultValue) {
         }
         return uuid;
       case DATE:
-        return parseDateValue(type, defaultValue);
+        Preconditions.checkArgument(
+            defaultValue.isTextual(), "Cannot parse default as a %s value: %s", type, defaultValue);
+        return DateTimeUtil.isoDateToDays(defaultValue.textValue());
       case TIME:
-        return parseTimeValue(type, defaultValue);
+        Preconditions.checkArgument(
+            defaultValue.isTextual(), "Cannot parse default as a %s value: %s", type, defaultValue);
+        return DateTimeUtil.isoTimeToMicros(defaultValue.textValue());
       case TIMESTAMP:
-        return parseTimestampValue(type, defaultValue);
+        Preconditions.checkArgument(
+            defaultValue.isTextual(), "Cannot parse default as a %s value: %s", type, defaultValue);
+        if (((Types.TimestampType) type).shouldAdjustToUTC()) {
+          String timestampTz = defaultValue.textValue();
+          Preconditions.checkArgument(
+              DateTimeUtil.isUTCTimestamptz(timestampTz),
+              "Cannot parse default as a %s value: %s, offset must be +00:00",
+              type,
+              defaultValue);
+          return DateTimeUtil.isoTimestamptzToMicros(timestampTz);
+        } else {
+          return DateTimeUtil.isoTimestampToMicros(defaultValue.textValue());
+        }
       case TIMESTAMP_NANO:
-        return parseTimestampNanoValue(type, defaultValue);
-
+        Preconditions.checkArgument(
+            defaultValue.isTextual(), "Cannot parse default as a %s value: %s", type, defaultValue);
+        if (((Types.TimestampNanoType) type).shouldAdjustToUTC()) {
+          String timestampTzNano = defaultValue.textValue();
+          Preconditions.checkArgument(
+              DateTimeUtil.isUTCTimestamptz(timestampTzNano),
+              "Cannot parse default as a %s value: %s, offset must be +00:00",
+              type,
+              defaultValue);
+          return DateTimeUtil.isoTimestamptzToNanos(timestampTzNano);
+        } else {
+          return DateTimeUtil.isoTimestampToNanos(defaultValue.textValue());
+        }
       case FIXED:
         Preconditions.checkArgument(
             defaultValue.isTextual(), "Cannot parse default as a %s value: %s", type, defaultValue);
@@ -387,68 +414,4 @@ public static void toJson(Type type, Object defaultValue, JsonGenerator generato
         throw new UnsupportedOperationException(String.format("Type: %s is not supported", type));
     }
   }
-
-  private static Object parseDateValue(Type type, JsonNode value) {
-    if (value.isTextual()) {
-      return DateTimeUtil.isoDateToDays(value.textValue());
-    } else if (value.isIntegralNumber() && value.canConvertToInt()) {
-      return value.intValue();
-    } else {
-      throw new IllegalArgumentException(
-          String.format("Cannot parse default as a %s value: %s", type, value));
-    }
-  }
-
-  private static Object parseTimeValue(Type type, JsonNode value) {
-    if (value.isTextual()) {
-      return DateTimeUtil.isoTimeToMicros(value.textValue());
-    } else if (value.isIntegralNumber() && value.canConvertToLong()) {
-      return value.longValue();
-    } else {
-      throw new IllegalArgumentException(
-          String.format("Cannot parse default as a %s value: %s", type, value));
-    }
-  }
-
-  private static Object parseTimestampValue(Type type, JsonNode value) {
-    if (value.isTextual()) {
-      if (((Types.TimestampType) type).shouldAdjustToUTC()) {
-        String timestampTz = value.textValue();
-        Preconditions.checkArgument(
-            DateTimeUtil.isUTCTimestamptz(timestampTz),
-            "Cannot parse default as a %s value: %s, offset must be +00:00",
-            type,
-            value);
-        return DateTimeUtil.isoTimestamptzToMicros(timestampTz);
-      } else {
-        return DateTimeUtil.isoTimestampToMicros(value.textValue());
-      }
-    } else if (value.isIntegralNumber() && value.canConvertToLong()) {
-      return value.longValue();
-    } else {
-      throw new IllegalArgumentException(
-          String.format("Cannot parse default as a %s value: %s", type, value));
-    }
-  }
-
-  private static Object parseTimestampNanoValue(Type type, JsonNode value) {
-    if (value.isTextual()) {
-      if (((Types.TimestampNanoType) type).shouldAdjustToUTC()) {
-        String timestampTzNano = value.textValue();
-        Preconditions.checkArgument(
-            DateTimeUtil.isUTCTimestamptz(timestampTzNano),
-            "Cannot parse default as a %s value: %s, offset must be +00:00",
-            type,
-            value);
-        return DateTimeUtil.isoTimestamptzToNanos(timestampTzNano);
-      } else {
-        return DateTimeUtil.isoTimestampToNanos(value.textValue());
-      }
-    } else if (value.isIntegralNumber() && value.canConvertToLong()) {
-      return value.longValue();
-    } else {
-      throw new IllegalArgumentException(
-          String.format("Cannot parse default as a %s value: %s", type, value));
-    }
-  }
 }

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

@@ -660,7 +660,7 @@ public static PlanTableScanResponse planTableScan(
               .fromSnapshotInclusive(request.startSnapshotId())
               .toSnapshot(request.endSnapshotId());
 
-      configuredScan = configureScan(incrementalScan, request, incrementalScan.schema());
+      configuredScan = configureScan(incrementalScan, request);
     } else {
       // Regular table scan at a specific snapshot
       TableScan tableScan = table.newScan();
@@ -670,7 +670,7 @@ public static PlanTableScanResponse planTableScan(
       }
 
       // Apply filters and projections using common method
-      configuredScan = configureScan(tableScan, request, tableScan.schema());
+      configuredScan = configureScan(tableScan, request);
     }
 
     if (shouldPlanAsync.test(configuredScan)) {
@@ -776,18 +776,17 @@ static void clearPlanningState() {
    *
    * @param scan the scan to configure
    * @param request the plan table scan request containing filters and projections
-   * @param schema the table schema to use for type-aware filter deserialization
    * @param <T> the specific scan type (TableScan, IncrementalAppendScan, etc.)
    * @return the configured scan with filters and projections applied
    */
   private static <T extends Scan<T, FileScanTask, ?>> T configureScan(
-      T scan, PlanTableScanRequest request, Schema schema) {
+      T scan, PlanTableScanRequest request) {
     T configuredScan = scan;
 
     if (request.select() != null) {
       configuredScan = configuredScan.select(request.select());
     }
-    Expression filter = request.filter(schema);
+    Expression filter = request.filter(scan.schema());
     if (filter != null) {
       configuredScan = configuredScan.filter(filter);
     }