[SPARK-51316][PYTHON][FOLLOW-UP] Revert unrelated changes and mark mapInPandas/mapInArrow batched in byte size

### What changes were proposed in this pull request?

This PR is a followup of apache/spark#50096 that reverts unrelated changes and mark mapInPandas/mapInArrow batched in byte size

### Why are the changes needed?

To make the original change self-contained, and mark mapInPandas/mapInArrow batched in byte size to be consistent.

### Does this PR introduce _any_ user-facing change?

No, the main change has not been released out yet.

### How was this patch tested?

Manually.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #50111 from HyukjinKwon/SPARK-51316-followup.

Authored-by: Hyukjin Kwon <gurwls223@apache.org>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

Author: a0x8o

common/variant/src/main/java/org/apache/spark/types/variant/ShreddingUtils.java

@@ -19,6 +19,7 @@
 
 import java.math.BigDecimal;
 import java.util.ArrayList;
+import java.util.UUID;
 
 import static org.apache.spark.types.variant.VariantUtil.*;
 
@@ -37,6 +38,7 @@ public interface ShreddedRow {
     BigDecimal getDecimal(int ordinal, int precision, int scale);
     String getString(int ordinal);
     byte[] getBinary(int ordinal);
+    UUID getUuid(int ordinal);
     ShreddedRow getStruct(int ordinal, int numFields);
     ShreddedRow getArray(int ordinal);
     int numElements();
@@ -99,6 +101,8 @@ public static void rebuild(ShreddedRow row, byte[] metadata, VariantSchema schem
           builder.appendBoolean(row.getBoolean(typedIdx));
         } else if (scalar instanceof VariantSchema.BinaryType) {
           builder.appendBinary(row.getBinary(typedIdx));
+        } else if (scalar instanceof VariantSchema.UuidType) {
+          builder.appendUuid(row.getUuid(typedIdx));
         } else if (scalar instanceof VariantSchema.DecimalType) {
           VariantSchema.DecimalType dt = (VariantSchema.DecimalType) scalar;
           builder.appendDecimal(row.getDecimal(typedIdx, dt.precision, dt.scale));

common/variant/src/main/java/org/apache/spark/types/variant/Variant.java

@@ -33,6 +33,7 @@
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.Locale;
+import java.util.UUID;
 
 import static org.apache.spark.types.variant.VariantUtil.*;
 
@@ -123,6 +124,11 @@ public Type getType() {
     return VariantUtil.getType(value, pos);
   }
 
+  // Get a UUID value from the variant.
+  public UUID getUuid() {
+    return VariantUtil.getUuid(value, pos);
+  }
+
   // Get the number of object fields in the variant.
   // It is only legal to call it when `getType()` is `Type.OBJECT`.
   public int objectSize() {
@@ -333,6 +339,9 @@ static void toJsonImpl(byte[] value, byte[] metadata, int pos, StringBuilder sb,
       case BINARY:
         appendQuoted(sb, Base64.getEncoder().encodeToString(VariantUtil.getBinary(value, pos)));
         break;
+      case UUID:
+        appendQuoted(sb, VariantUtil.getUuid(value, pos).toString());
+        break;
     }
   }
 }

common/variant/src/main/java/org/apache/spark/types/variant/VariantBuilder.java

@@ -26,6 +26,8 @@
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.nio.charset.StandardCharsets;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.*;
 
 import com.fasterxml.jackson.core.JsonFactory;
@@ -240,6 +242,18 @@ public void appendBinary(byte[] binary) {
     writePos += binary.length;
   }
 
+  public void appendUuid(UUID uuid) {
+    checkCapacity(1 + 16);
+    writeBuffer[writePos++] = primitiveHeader(UUID);
+
+    // UUID is stored big-endian, so don't use writeLong.
+    ByteBuffer buffer = ByteBuffer.wrap(writeBuffer, writePos, 16);
+    buffer.order(ByteOrder.BIG_ENDIAN);
+    buffer.putLong(writePos, uuid.getMostSignificantBits());
+    buffer.putLong(writePos + 8, uuid.getLeastSignificantBits());
+    writePos += 16;
+  }
+
   // Add a key to the variant dictionary. If the key already exists, the dictionary is not modified.
   // In either case, return the id of the key.
   public int addKey(String key) {

common/variant/src/main/java/org/apache/spark/types/variant/VariantSchema.java

@@ -99,6 +99,9 @@ public static final class TimestampType extends ScalarType {
   public static final class TimestampNTZType extends ScalarType {
   }
 
+  public static final class UuidType extends ScalarType {
+  }
+
   // The index of the typed_value, value, and metadata fields in the schema, respectively. If a
   // given field is not in the schema, its value must be set to -1 to indicate that it is invalid.
   // The indices of valid fields should be contiguous and start from 0.

common/variant/src/main/java/org/apache/spark/types/variant/VariantShreddingWriter.java

@@ -283,6 +283,11 @@ private static Object tryTypedShred(
           return v.getBinary();
         }
         break;
+      case UUID:
+        if (targetType instanceof VariantSchema.UuidType) {
+          return v.getUuid();
+        }
+        break;
     }
     // The stored type does not match the requested shredding type. Return null, and the caller
     // will store the result in untyped_value.

common/variant/src/main/java/org/apache/spark/types/variant/VariantUtil.java

@@ -23,7 +23,10 @@
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.Arrays;
+import java.util.UUID;
 
 /**
  * This class defines constants related to the variant format and provides functions for
@@ -121,6 +124,9 @@ public class VariantUtil {
   // string size) + (size bytes of string content).
   public static final int LONG_STR = 16;
 
+  // UUID, 16-byte big-endian.
+  public static final int UUID = 20;
+
   public static final byte VERSION = 1;
   // The lower 4 bits of the first metadata byte contain the version.
   public static final byte VERSION_MASK = 0x0F;
@@ -239,6 +245,7 @@ public enum Type {
     TIMESTAMP_NTZ,
     FLOAT,
     BINARY,
+    UUID,
   }
 
   public static int getTypeInfo(byte[] value, int pos) {
@@ -291,6 +298,8 @@ public static Type getType(byte[] value, int pos) {
             return Type.BINARY;
           case LONG_STR:
             return Type.STRING;
+          case UUID:
+            return Type.UUID;
           default:
             throw unknownPrimitiveTypeInVariant(typeInfo);
         }
@@ -342,6 +351,8 @@ public static int valueSize(byte[] value, int pos) {
           case BINARY:
           case LONG_STR:
             return 1 + U32_SIZE + readUnsigned(value, pos + 1, U32_SIZE);
+          case UUID:
+            return 17;
           default:
             throw unknownPrimitiveTypeInVariant(typeInfo);
         }
@@ -497,6 +508,20 @@ public static String getString(byte[] value, int pos) {
     throw unexpectedType(Type.STRING);
   }
 
+  // Get a UUID value from variant value `value[pos...]`.
+  // Throw `MALFORMED_VARIANT` if the variant is malformed.
+  public static UUID getUuid(byte[] value, int pos) {
+    checkIndex(pos, value.length);
+    int basicType = value[pos] & BASIC_TYPE_MASK;
+    int typeInfo = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+    if (basicType != PRIMITIVE || typeInfo != UUID) throw unexpectedType(Type.UUID);
+    int start = pos + 1;
+    checkIndex(start + 15, value.length);
+    // UUID values are big-endian, so we can't use VariantUtil.readLong().
+    ByteBuffer bb = ByteBuffer.wrap(value, start, 16).order(ByteOrder.BIG_ENDIAN);
+    return new UUID(bb.getLong(), bb.getLong());
+  }
+
   public interface ObjectHandler<T> {
     /**
      * @param size Number of object fields.

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/PostgresIntegrationSuite.scala

@@ -185,6 +185,9 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JDBCT
       "('amy', '2022-05-19', '2022-05-19 00:00:00')").executeUpdate()
     connection.prepareStatement("INSERT INTO datetime VALUES " +
       "('alex', '2022-05-18', '2022-05-18 00:00:00')").executeUpdate()
+    // '2022-01-01' is Saturday and is in ISO year 2021.
+    connection.prepareStatement("INSERT INTO datetime VALUES " +
+      "('tom', '2022-01-01', '2022-01-01 00:00:00')").executeUpdate()
   }
 
   override def testUpdateColumnType(tbl: String): Unit = {
@@ -279,17 +282,19 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JDBCT
     val df4 = sql(s"SELECT name FROM $tbl WHERE hour(time1) = 0 AND minute(time1) = 0")
     checkFilterPushed(df4)
     val rows4 = df4.collect()
-    assert(rows4.length === 2)
+    assert(rows4.length === 3)
     assert(rows4(0).getString(0) === "amy")
     assert(rows4(1).getString(0) === "alex")
+    assert(rows4(2).getString(0) === "tom")
 
     val df5 = sql(s"SELECT name FROM $tbl WHERE " +
-      "extract(WEEk from date1) > 10 AND extract(YEAROFWEEK from date1) = 2022")
+      "extract(WEEK from date1) > 10 AND extract(YEAR from date1) = 2022")
     checkFilterPushed(df5)
     val rows5 = df5.collect()
-    assert(rows5.length === 2)
+    assert(rows5.length === 3)
     assert(rows5(0).getString(0) === "amy")
     assert(rows5(1).getString(0) === "alex")
+    assert(rows5(2).getString(0) === "tom")
 
     val df6 = sql(s"SELECT name FROM $tbl WHERE date_add(date1, 1) = date'2022-05-20' " +
       "AND datediff(date1, '2022-05-10') > 0")
@@ -304,11 +309,25 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JDBCT
     assert(rows7.length === 1)
     assert(rows7(0).getString(0) === "alex")
 
-    val df8 = sql(s"SELECT name FROM $tbl WHERE dayofweek(date1) = 4")
-    checkFilterPushed(df8)
-    val rows8 = df8.collect()
-    assert(rows8.length === 1)
-    assert(rows8(0).getString(0) === "alex")
+    withClue("dayofweek") {
+      val dow = sql(s"SELECT dayofweek(date1) FROM $tbl WHERE name = 'alex'")
+        .collect().head.getInt(0)
+      val df = sql(s"SELECT name FROM $tbl WHERE dayofweek(date1) = $dow")
+      checkFilterPushed(df)
+      val rows = df.collect()
+      assert(rows.length === 1)
+      assert(rows(0).getString(0) === "alex")
+    }
+
+    withClue("yearofweek") {
+      val yow = sql(s"SELECT extract(YEAROFWEEK from date1) FROM $tbl WHERE name = 'tom'")
+        .collect().head.getInt(0)
+      val df = sql(s"SELECT name FROM $tbl WHERE extract(YEAROFWEEK from date1) = $yow")
+      checkFilterPushed(df)
+      val rows = df.collect()
+      assert(rows.length === 1)
+      assert(rows(0).getString(0) === "tom")
+    }
 
     val df9 = sql(s"SELECT name FROM $tbl WHERE " +
       "dayofyear(date1) > 100 order by dayofyear(date1) limit 1")

python/docs/source/development/logger.rst

@@ -57,7 +57,14 @@ Example log entry:
       "exception": {
         "class": "Py4JJavaError",
         "msg": "An error occurred while calling o52.showString.\n: org.apache.spark.SparkArithmeticException: [DIVIDE_BY_ZERO] Division by zero. Use `try_divide` to tolerate divisor being 0 and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error. SQLSTATE: 22012\n== DataFrame ==\n\"divide\" was called from\n/path/to/file.py:17 ...",
-        "stacktrace": ["Traceback (most recent call last):", "  File \".../spark/python/pyspark/errors/exceptions/captured.py\", line 247, in deco", "    return f(*a, **kw)", "  File \".../lib/python3.9/site-packages/py4j/protocol.py\", line 326, in get_return_value" ...]
+        "stacktrace": [
+          {
+            "class": null,
+            "method": "deco",
+            "file": ".../spark/python/pyspark/errors/exceptions/captured.py",
+            "line": "247"
+          }
+        ]
       },
     }
 

python/pyspark/sql/connect/column.py

@@ -44,7 +44,6 @@
     Expression,
     UnresolvedFunction,
     UnresolvedExtractValue,
-    LazyExpression,
     LiteralExpression,
     CaseWhen,
     SortOrder,
@@ -459,7 +458,7 @@ def over(self, window: "WindowSpec") -> ParentColumn:  # type: ignore[override]
         return Column(WindowExpression(windowFunction=self._expr, windowSpec=window))
 
     def outer(self) -> ParentColumn:
-        return Column(LazyExpression(self._expr))
+        return Column(self._expr)
 
     def isin(self, *cols: Any) -> ParentColumn:
         if len(cols) == 1 and isinstance(cols[0], (list, set)):

python/pyspark/sql/connect/expressions.py

@@ -1233,25 +1233,6 @@ def __repr__(self) -> str:
         return f"{self._key} => {self._value}"
 
 
-class LazyExpression(Expression):
-    def __init__(self, expr: Expression):
-        assert isinstance(expr, Expression)
-        super().__init__()
-        self._expr = expr
-
-    def to_plan(self, session: "SparkConnectClient") -> proto.Expression:
-        expr = self._create_proto_expression()
-        expr.lazy_expression.child.CopyFrom(self._expr.to_plan(session))
-        return expr
-
-    @property
-    def children(self) -> Sequence["Expression"]:
-        return [self._expr]
-
-    def __repr__(self) -> str:
-        return f"lazy({self._expr})"
-
-
 class SubqueryExpression(Expression):
     def __init__(
         self,