MaxGekk commented on code in PR #57089:
URL: https://github.com/apache/spark/pull/57089#discussion_r3574276154


##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Overlaps.scala:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, 
ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.Block.BlockHelper
+import org.apache.spark.sql.catalyst.trees.QuaternaryLike
+import org.apache.spark.sql.catalyst.trees.TreePattern._
+import org.apache.spark.sql.types.{BooleanType, DataType, DateType, 
TimestampNTZType, TimestampType, TimeType}
+
+/**
+ * Implements the ANSI SQL OVERLAPS predicate for datetime periods.
+ *
+ * Syntax:
+ *   (start1, end1) OVERLAPS (start2, end2)
+ *
+ * Semantics (per ISO/IEC 9075-2):
+ * 1. Each period is normalized so start <= end (endpoints swapped if needed).
+ * 2. A zero-length period represents a single point in time.
+ * 3. Two periods overlap iff they share at least one common point:
+ *    normalizedStart1 < normalizedEnd2 AND normalizedStart2 < normalizedEnd1
+ *    (For zero-length periods, use <= for the point-containment check.)
+ * 4. NULL endpoints follow standard three-valued logic.
+ */
+case class Overlaps(
+    start1: Expression,
+    end1: Expression,
+    start2: Expression,
+    end2: Expression)
+  extends Expression with QuaternaryLike[Expression] {
+
+  override def first: Expression = start1
+  override def second: Expression = end1
+  override def third: Expression = start2
+  override def fourth: Expression = end2
+
+  override val nodePatterns: Seq[TreePattern] = Seq(OVERLAPS)
+
+  override def dataType: DataType = BooleanType
+  override def nullable: Boolean = start1.nullable || end1.nullable ||
+    start2.nullable || end2.nullable
+
+  // All four endpoints must be the same datetime type.
+  // The (start, interval) form is supported: the interval is added to the 
start
+  // to produce the endpoint. This is handled transparently during analysis via
+  // the analyzer rewriting Overlaps nodes with interval children.
+  override def checkInputDataTypes(): TypeCheckResult = {
+    val types = Seq(start1.dataType, end1.dataType, start2.dataType, 
end2.dataType)
+
+    // Check all are the same datetime family (after interval resolution)
+    val distinctTypes = types.map(canonicalType).distinct
+    if (distinctTypes.length != 1) {
+      TypeCheckResult.TypeCheckFailure(
+        s"All endpoints in OVERLAPS must be the same datetime type, " +

Review Comment:
   New expressions generally use 
`TypeCheckResult.DataTypeMismatch(errorSubClass, messageParameters)` (backed by 
an entry in `error-conditions.json`) rather than the free-text 
`TypeCheckFailure(String)`. As written, these surface under the generic 
`DATATYPE_MISMATCH.TYPE_CHECK_FAILURE_WITH_HINT` class (I confirmed by running 
a mixed-type query), so the message isn't parameterized or localizable like 
peer expressions' type errors. Consider a dedicated sub-class (e.g. an 
`OVERLAPS`-specific `DATATYPE_MISMATCH` variant), or reuse an existing 
datetime-type-mismatch sub-class.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Overlaps.scala:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, 
ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.Block.BlockHelper
+import org.apache.spark.sql.catalyst.trees.QuaternaryLike
+import org.apache.spark.sql.catalyst.trees.TreePattern._
+import org.apache.spark.sql.types.{BooleanType, DataType, DateType, 
TimestampNTZType, TimestampType, TimeType}
+
+/**
+ * Implements the ANSI SQL OVERLAPS predicate for datetime periods.
+ *
+ * Syntax:
+ *   (start1, end1) OVERLAPS (start2, end2)
+ *
+ * Semantics (per ISO/IEC 9075-2):
+ * 1. Each period is normalized so start <= end (endpoints swapped if needed).
+ * 2. A zero-length period represents a single point in time.
+ * 3. Two periods overlap iff they share at least one common point:
+ *    normalizedStart1 < normalizedEnd2 AND normalizedStart2 < normalizedEnd1
+ *    (For zero-length periods, use <= for the point-containment check.)
+ * 4. NULL endpoints follow standard three-valued logic.
+ */
+case class Overlaps(
+    start1: Expression,
+    end1: Expression,
+    start2: Expression,
+    end2: Expression)
+  extends Expression with QuaternaryLike[Expression] {
+
+  override def first: Expression = start1
+  override def second: Expression = end1
+  override def third: Expression = start2
+  override def fourth: Expression = end2
+
+  override val nodePatterns: Seq[TreePattern] = Seq(OVERLAPS)
+
+  override def dataType: DataType = BooleanType
+  override def nullable: Boolean = start1.nullable || end1.nullable ||
+    start2.nullable || end2.nullable
+
+  // All four endpoints must be the same datetime type.
+  // The (start, interval) form is supported: the interval is added to the 
start
+  // to produce the endpoint. This is handled transparently during analysis via
+  // the analyzer rewriting Overlaps nodes with interval children.
+  override def checkInputDataTypes(): TypeCheckResult = {
+    val types = Seq(start1.dataType, end1.dataType, start2.dataType, 
end2.dataType)
+
+    // Check all are the same datetime family (after interval resolution)
+    val distinctTypes = types.map(canonicalType).distinct
+    if (distinctTypes.length != 1) {
+      TypeCheckResult.TypeCheckFailure(
+        s"All endpoints in OVERLAPS must be the same datetime type, " +
+          s"but got ${types.map(_.sql).mkString(", ")}")
+    } else if (!isSupportedType(distinctTypes.head)) {
+      TypeCheckResult.TypeCheckFailure(
+        s"OVERLAPS requires datetime endpoints (TIME, DATE, TIMESTAMP, or 
TIMESTAMP_NTZ), " +
+          s"but got ${distinctTypes.head.sql}")
+    } else {
+      TypeCheckResult.TypeCheckSuccess
+    }
+  }
+
+  private def canonicalType(dt: DataType): DataType = dt match {
+    case _: TimeType => TimeType(6) // normalize precision for comparison
+    case other => other
+  }
+
+  private def isSupportedType(dt: DataType): Boolean = dt match {
+    case _: TimeType | DateType | TimestampType | TimestampNTZType => true
+    case _ => false
+  }
+
+  override def foldable: Boolean = children.forall(_.foldable)
+
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): 
ExprCode = {
+    // Fall back to interpreted evaluation

Review Comment:
   This hand-rolls exactly what the standard `CodegenFallback` trait provides 
(`addReferenceObj`/`references` + `eval` + null-unwrap), but less completely — 
`CodegenFallback.doGenCode` also walks children and calls 
`initialize(partitionIndex)` on any `Nondeterministic` child, which this 
version skips. If an endpoint ever contains a nondeterministic expression, it 
wouldn't be initialized under whole-stage codegen.
   
   Suggest `extends Expression with QuaternaryLike[Expression] with 
CodegenFallback` and deleting this `doGenCode` entirely — the trait handles the 
fallback correctly.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Overlaps.scala:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, 
ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.Block.BlockHelper
+import org.apache.spark.sql.catalyst.trees.QuaternaryLike
+import org.apache.spark.sql.catalyst.trees.TreePattern._
+import org.apache.spark.sql.types.{BooleanType, DataType, DateType, 
TimestampNTZType, TimestampType, TimeType}
+
+/**
+ * Implements the ANSI SQL OVERLAPS predicate for datetime periods.
+ *
+ * Syntax:
+ *   (start1, end1) OVERLAPS (start2, end2)
+ *
+ * Semantics (per ISO/IEC 9075-2):
+ * 1. Each period is normalized so start <= end (endpoints swapped if needed).
+ * 2. A zero-length period represents a single point in time.
+ * 3. Two periods overlap iff they share at least one common point:
+ *    normalizedStart1 < normalizedEnd2 AND normalizedStart2 < normalizedEnd1
+ *    (For zero-length periods, use <= for the point-containment check.)
+ * 4. NULL endpoints follow standard three-valued logic.
+ */
+case class Overlaps(
+    start1: Expression,
+    end1: Expression,
+    start2: Expression,
+    end2: Expression)
+  extends Expression with QuaternaryLike[Expression] {
+
+  override def first: Expression = start1
+  override def second: Expression = end1
+  override def third: Expression = start2
+  override def fourth: Expression = end2
+
+  override val nodePatterns: Seq[TreePattern] = Seq(OVERLAPS)
+
+  override def dataType: DataType = BooleanType
+  override def nullable: Boolean = start1.nullable || end1.nullable ||
+    start2.nullable || end2.nullable
+
+  // All four endpoints must be the same datetime type.
+  // The (start, interval) form is supported: the interval is added to the 
start
+  // to produce the endpoint. This is handled transparently during analysis via
+  // the analyzer rewriting Overlaps nodes with interval children.
+  override def checkInputDataTypes(): TypeCheckResult = {
+    val types = Seq(start1.dataType, end1.dataType, start2.dataType, 
end2.dataType)
+
+    // Check all are the same datetime family (after interval resolution)
+    val distinctTypes = types.map(canonicalType).distinct
+    if (distinctTypes.length != 1) {
+      TypeCheckResult.TypeCheckFailure(
+        s"All endpoints in OVERLAPS must be the same datetime type, " +
+          s"but got ${types.map(_.sql).mkString(", ")}")
+    } else if (!isSupportedType(distinctTypes.head)) {
+      TypeCheckResult.TypeCheckFailure(
+        s"OVERLAPS requires datetime endpoints (TIME, DATE, TIMESTAMP, or 
TIMESTAMP_NTZ), " +
+          s"but got ${distinctTypes.head.sql}")
+    } else {
+      TypeCheckResult.TypeCheckSuccess
+    }
+  }
+
+  private def canonicalType(dt: DataType): DataType = dt match {
+    case _: TimeType => TimeType(6) // normalize precision for comparison
+    case other => other
+  }
+
+  private def isSupportedType(dt: DataType): Boolean = dt match {
+    case _: TimeType | DateType | TimestampType | TimestampNTZType => true
+    case _ => false
+  }
+
+  override def foldable: Boolean = children.forall(_.foldable)
+
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): 
ExprCode = {
+    // Fall back to interpreted evaluation
+    val thisTerm = ctx.addReferenceObj("overlaps", this)
+    val inputTerm = ctx.INPUT_ROW
+    ev.copy(code =
+      code"""
+        Object ${ev.value}Obj = $thisTerm.eval($inputTerm);
+        boolean ${ev.isNull} = ${ev.value}Obj == null;
+        boolean ${ev.value} = ${ev.isNull} ? false : (Boolean) ${ev.value}Obj;
+      """)
+  }
+
+  override def eval(input: InternalRow): Any = {
+    val s1 = start1.eval(input)
+    val e1 = end1.eval(input)
+    val s2 = start2.eval(input)
+    val e2 = end2.eval(input)
+
+    // NULL handling: if any endpoint is null, result is null
+    if (s1 == null || e1 == null || s2 == null || e2 == null) {
+      return null
+    }
+
+    start1.dataType match {
+      case DateType =>
+        overlapCheck(s1.asInstanceOf[Int], e1.asInstanceOf[Int],
+          s2.asInstanceOf[Int], e2.asInstanceOf[Int])
+      case _: TimeType | TimestampType | TimestampNTZType =>
+        overlapCheck(s1.asInstanceOf[Long], e1.asInstanceOf[Long],
+          s2.asInstanceOf[Long], e2.asInstanceOf[Long])
+      case _ => null

Review Comment:
   This arm is unreachable after `checkInputDataTypes` (which requires a 
supported datetime type), but returning `null` for a `Boolean` predicate would 
silently mask a future type-check regression. Prefer `throw 
SparkException.internalError(...)` (or a `QueryExecutionErrors` helper) so an 
unexpected type fails loudly instead of yielding a wrong NULL.



##########
sql/core/src/test/resources/sql-tests/inputs/overlaps.sql:
##########
@@ -0,0 +1,53 @@
+-- Test the OVERLAPS predicate for datetime periods
+
+-- TIME overlapping periods
+SELECT (TIME'09:00:00', TIME'12:00:00') OVERLAPS (TIME'11:00:00', 
TIME'13:00:00');
+
+-- TIME non-overlapping periods
+SELECT (TIME'09:00:00', TIME'11:00:00') OVERLAPS (TIME'13:00:00', 
TIME'15:00:00');
+
+-- TIME touching endpoints (not overlapping per ANSI)
+SELECT (TIME'09:00:00', TIME'12:00:00') OVERLAPS (TIME'12:00:00', 
TIME'13:00:00');
+
+-- TIME endpoint normalization (swapped start/end)
+SELECT (TIME'12:00:00', TIME'09:00:00') OVERLAPS (TIME'11:00:00', 
TIME'13:00:00');
+
+-- TIME zero-length period (point) contained in period
+SELECT (TIME'10:00:00', TIME'10:00:00') OVERLAPS (TIME'09:00:00', 
TIME'11:00:00');
+
+-- TIME zero-length period (point) at boundary (not overlapping)
+SELECT (TIME'12:00:00', TIME'12:00:00') OVERLAPS (TIME'09:00:00', 
TIME'12:00:00');
+
+-- TIME two identical points overlap
+SELECT (TIME'10:00:00', TIME'10:00:00') OVERLAPS (TIME'10:00:00', 
TIME'10:00:00');
+
+-- TIME two different points do not overlap
+SELECT (TIME'10:00:00', TIME'10:00:00') OVERLAPS (TIME'11:00:00', 
TIME'11:00:00');
+
+-- TIME with microsecond precision
+SELECT (TIME'09:00:00.000001', TIME'12:00:00.999999') OVERLAPS 
(TIME'12:00:00.999998', TIME'13:00:00');
+
+-- TIME with interval form (explicit addition)
+SELECT (TIME'09:00:00', TIME'09:00:00' + INTERVAL '3' HOUR) OVERLAPS 
(TIME'11:00:00', TIME'13:00:00');
+
+-- TIME with interval form (non-overlapping, explicit addition)
+SELECT (TIME'09:00:00', TIME'09:00:00' + INTERVAL '1' HOUR) OVERLAPS 
(TIME'11:00:00', TIME'13:00:00');
+
+-- TIME with raw interval endpoint (resolved by ResolveOverlaps rule)
+SELECT (TIME'09:00:00', INTERVAL '3' HOUR) OVERLAPS (TIME'11:00:00', 
TIME'13:00:00');
+
+-- NULL endpoints (typed NULL from CAST)
+SELECT (TIME'09:00:00', CAST(NULL AS TIME)) OVERLAPS (TIME'11:00:00', 
TIME'13:00:00');
+SELECT (CAST(NULL AS TIME), TIME'12:00:00') OVERLAPS (TIME'11:00:00', 
TIME'13:00:00');
+
+-- DATE overlapping periods
+SELECT (DATE'2024-01-01', DATE'2024-06-30') OVERLAPS (DATE'2024-03-01', 
DATE'2024-12-31');
+
+-- DATE non-overlapping periods
+SELECT (DATE'2024-01-01', DATE'2024-03-01') OVERLAPS (DATE'2024-06-01', 
DATE'2024-12-31');
+
+-- TIMESTAMP overlapping periods
+SELECT (TIMESTAMP'2024-01-01 09:00:00', TIMESTAMP'2024-01-01 12:00:00') 
OVERLAPS (TIMESTAMP'2024-01-01 11:00:00', TIMESTAMP'2024-01-01 13:00:00');
+
+-- TIMESTAMP_NTZ overlapping periods

Review Comment:
   The golden coverage is thorough on the happy path, but every query here is 
`SELECT <literals>`. Consider adding cases for the negative/edge paths — I 
verified these all behave correctly, they just aren't pinned:
   - mixed-type rejection, e.g. `(DATE, DATE) OVERLAPS (TIMESTAMP, TIMESTAMP)` 
→ `DATATYPE_MISMATCH`
   - non-datetime rejection, e.g. `(1, 5) OVERLAPS (3, 7)` → `DATATYPE_MISMATCH`
   - TIME interval overflow, e.g. `(TIME'23:00:00', INTERVAL '3' HOUR) OVERLAPS 
(...)` → `DATETIME_OVERFLOW`
   - a `WHERE (c1, c2) OVERLAPS (c3, c4)` over a small table — this exercises 
the codegen fallback inside a filter, which the literal `SELECT`s don't.



##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Overlaps.scala:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, 
ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.Block.BlockHelper
+import org.apache.spark.sql.catalyst.trees.QuaternaryLike
+import org.apache.spark.sql.catalyst.trees.TreePattern._
+import org.apache.spark.sql.types.{BooleanType, DataType, DateType, 
TimestampNTZType, TimestampType, TimeType}
+
+/**
+ * Implements the ANSI SQL OVERLAPS predicate for datetime periods.
+ *
+ * Syntax:
+ *   (start1, end1) OVERLAPS (start2, end2)
+ *
+ * Semantics (per ISO/IEC 9075-2):
+ * 1. Each period is normalized so start <= end (endpoints swapped if needed).
+ * 2. A zero-length period represents a single point in time.
+ * 3. Two periods overlap iff they share at least one common point:
+ *    normalizedStart1 < normalizedEnd2 AND normalizedStart2 < normalizedEnd1
+ *    (For zero-length periods, use <= for the point-containment check.)
+ * 4. NULL endpoints follow standard three-valued logic.
+ */
+case class Overlaps(
+    start1: Expression,
+    end1: Expression,
+    start2: Expression,
+    end2: Expression)
+  extends Expression with QuaternaryLike[Expression] {
+
+  override def first: Expression = start1
+  override def second: Expression = end1
+  override def third: Expression = start2
+  override def fourth: Expression = end2
+
+  override val nodePatterns: Seq[TreePattern] = Seq(OVERLAPS)
+
+  override def dataType: DataType = BooleanType
+  override def nullable: Boolean = start1.nullable || end1.nullable ||
+    start2.nullable || end2.nullable
+
+  // All four endpoints must be the same datetime type.
+  // The (start, interval) form is supported: the interval is added to the 
start
+  // to produce the endpoint. This is handled transparently during analysis via
+  // the analyzer rewriting Overlaps nodes with interval children.
+  override def checkInputDataTypes(): TypeCheckResult = {
+    val types = Seq(start1.dataType, end1.dataType, start2.dataType, 
end2.dataType)
+
+    // Check all are the same datetime family (after interval resolution)
+    val distinctTypes = types.map(canonicalType).distinct
+    if (distinctTypes.length != 1) {
+      TypeCheckResult.TypeCheckFailure(
+        s"All endpoints in OVERLAPS must be the same datetime type, " +
+          s"but got ${types.map(_.sql).mkString(", ")}")
+    } else if (!isSupportedType(distinctTypes.head)) {
+      TypeCheckResult.TypeCheckFailure(
+        s"OVERLAPS requires datetime endpoints (TIME, DATE, TIMESTAMP, or 
TIMESTAMP_NTZ), " +
+          s"but got ${distinctTypes.head.sql}")
+    } else {
+      TypeCheckResult.TypeCheckSuccess
+    }
+  }
+
+  private def canonicalType(dt: DataType): DataType = dt match {
+    case _: TimeType => TimeType(6) // normalize precision for comparison
+    case other => other
+  }
+
+  private def isSupportedType(dt: DataType): Boolean = dt match {
+    case _: TimeType | DateType | TimestampType | TimestampNTZType => true
+    case _ => false
+  }
+
+  override def foldable: Boolean = children.forall(_.foldable)
+
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): 
ExprCode = {
+    // Fall back to interpreted evaluation
+    val thisTerm = ctx.addReferenceObj("overlaps", this)
+    val inputTerm = ctx.INPUT_ROW
+    ev.copy(code =
+      code"""
+        Object ${ev.value}Obj = $thisTerm.eval($inputTerm);
+        boolean ${ev.isNull} = ${ev.value}Obj == null;
+        boolean ${ev.value} = ${ev.isNull} ? false : (Boolean) ${ev.value}Obj;
+      """)
+  }
+
+  override def eval(input: InternalRow): Any = {
+    val s1 = start1.eval(input)
+    val e1 = end1.eval(input)
+    val s2 = start2.eval(input)
+    val e2 = end2.eval(input)
+
+    // NULL handling: if any endpoint is null, result is null
+    if (s1 == null || e1 == null || s2 == null || e2 == null) {
+      return null
+    }
+
+    start1.dataType match {
+      case DateType =>
+        overlapCheck(s1.asInstanceOf[Int], e1.asInstanceOf[Int],
+          s2.asInstanceOf[Int], e2.asInstanceOf[Int])
+      case _: TimeType | TimestampType | TimestampNTZType =>
+        overlapCheck(s1.asInstanceOf[Long], e1.asInstanceOf[Long],
+          s2.asInstanceOf[Long], e2.asInstanceOf[Long])
+      case _ => null
+    }
+  }
+
+  private def overlapCheck(s1: Long, e1: Long, s2: Long, e2: Long): Boolean = {
+    // Normalize: ensure start <= end
+    val (ns1, ne1) = if (s1 <= e1) (s1, e1) else (e1, s1)
+    val (ns2, ne2) = if (s2 <= e2) (s2, e2) else (e2, s2)
+
+    val isPoint1 = ns1 == ne1
+    val isPoint2 = ns2 == ne2
+
+    if (isPoint1 && isPoint2) {
+      ns1 == ns2
+    } else if (isPoint1) {
+      ns1 >= ns2 && ns1 < ne2
+    } else if (isPoint2) {
+      ns2 >= ns1 && ns2 < ne1
+    } else {
+      ns1 < ne2 && ns2 < ne1
+    }
+  }
+
+  private def overlapCheck(s1: Int, e1: Int, s2: Int, e2: Int): Boolean = {

Review Comment:
   This `Int` overload is byte-for-byte identical to the `Long` one above 
except the parameter type. Since DATE is the only `Int`-backed case and widens 
losslessly to `Long`, you could drop this overload and call the `Long` version 
from the `DateType` branch in `eval` (widening the four ints), collapsing two 
identical bodies into one.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to