Expression Operations and Operators

trait ImplicitExpressionOperations {
  private[flink] def expr: Expression
  
  // Field aliasing with symbols
  def as(name: Symbol, extraNames: Symbol*): Expression
  
  // Comparison operators  
  def >(other: Expression): Expression
  def >=(other: Expression): Expression
  def <(other: Expression): Expression
  def <=(other: Expression): Expression
  def ===(other: Expression): Expression
  def !==(other: Expression): Expression
  
  // Logical operators
  def &&(other: Expression): Expression
  def ||(other: Expression): Expression
  def unary_!: Expression
  
  // Arithmetic operators
  def +(other: Expression): Expression
  def -(other: Expression): Expression
  def *(other: Expression): Expression
  def /(other: Expression): Expression
  def %(other: Expression): Expression
  def unary_-: Expression
  def unary_+: Expression
  
  // Specialized operations
  def to(other: Expression): Expression
  def ?(ifTrue: Expression, ifFalse: Expression): Expression
  def rows: Expression
}

def as(name: Symbol, extraNames: Symbol*): Expression

import org.apache.flink.table.api._

// Single alias
val aliased = $"price" * 1.1 as 'totalPrice

// Multiple aliases (for expressions that expand to multiple fields)
val multiAlias = someComplexExpression as ('field1, 'field2, 'field3)

// String-based aliasing (inherited)
val stringAlias = $"price" * 1.1 as "totalPrice"

def >(other: Expression): Expression          // Greater than
def >=(other: Expression): Expression         // Greater than or equal
def <(other: Expression): Expression          // Less than
def <=(other: Expression): Expression         // Less than or equal
def ===(other: Expression): Expression        // Equal (use === to avoid conflicts)
def !==(other: Expression): Expression        // Not equal

// Numeric comparisons
val condition1 = $"age" > lit(18)
val condition2 = $"salary" >= lit(50000.0)
val condition3 = $"score" < lit(100)
val condition4 = $"rating" <= lit(5.0)

// Equality comparisons
val isActive = $"status" === lit("ACTIVE")
val notDeleted = $"deleted" !== lit(true)

// String comparisons
val nameFilter = $"name" === lit("John")

// Date comparisons
val recentOrders = $"orderDate" > lit(LocalDate.now().minusDays(30))

def &&(other: Expression): Expression         // Logical AND
def ||(other: Expression): Expression         // Logical OR  
def unary_!: Expression                       // Logical NOT

// Complex conditions
val adultActive = ($"age" >= lit(18)) && ($"status" === lit("ACTIVE"))
val eligibleUser = ($"premium" === lit(true)) || ($"trialExpiry" > currentDate())
val nonDeleted = !($"deleted" === lit(true))

// Chained logical operations
val complexFilter = ($"category" === lit("ELECTRONICS")) && 
                   (($"price" > lit(100)) || ($"discount" > lit(0.2))) &&
                   !($"outOfStock" === lit(true))

def +(other: Expression): Expression          // Addition
def -(other: Expression): Expression          // Subtraction
def *(other: Expression): Expression          // Multiplication
def /(other: Expression): Expression          // Division
def %(other: Expression): Expression          // Modulo
def unary_-: Expression                       // Negation
def unary_+: Expression                       // Positive (identity)

// Basic arithmetic
val totalPrice = $"price" + $"tax"
val discount = $"originalPrice" - $"finalPrice"
val area = $"width" * $"height"
val average = $"sum" / $"count"
val remainder = $"value" % lit(10)

// Unary operations
val negated = -$"balance"
val positive = +$"amount"

// Complex calculations
val finalAmount = ($"price" * $"quantity") + 
                 ($"price" * $"quantity" * $"taxRate") - 
                 $"discount"

// Mathematical expressions with literals
val adjustedScore = ($"score" * lit(1.1)) + lit(5.0)

def to(other: Expression): Expression

// Select columns from 'start' to 'end'
val columnRange = $"col1" to $"col5"  // Selects col1, col2, col3, col4, col5

// Usage in select
val result = table.select(columnRange)

def ?(ifTrue: Expression, ifFalse: Expression): Expression

// Conditional expressions
val status = ($"age" >= lit(18)) ? lit("Adult") : lit("Minor")
val category = ($"score" > lit(80)) ? lit("High") : 
              (($"score" > lit(60)) ? lit("Medium") : lit("Low"))

// Null handling
val displayName = $"nickname".isNotNull ? $"nickname" : $"firstName"

// Numeric conditionals
val finalPrice = ($"memberType" === lit("PREMIUM")) ? 
                ($"price" * lit(0.9)) : $"price"

def rows: Expression

// Window with row-based bounds
val windowSpec = Over
  .partitionBy($"userId")
  .orderBy($"timestamp")
  .preceding(lit(5).rows)  // 5 rows preceding
  .currentRow()

// Usage in window function
val result = table.select(
  $"*",
  $"amount".sum.over(windowSpec) as "runningSum"
)

// Multi-level conditionals
val priorityLevel = ($"urgent" === lit(true)) ? lit("HIGH") :
                   (($"important" === lit(true)) ? lit("MEDIUM") : lit("LOW"))

// Combining with arithmetic
val adjustedPrice = ($"customerType" === lit("VIP")) ? 
                   ($"price" * lit(0.8)) :
                   (($"customerType" === lit("MEMBER")) ? 
                    ($"price" * lit(0.9)) : $"price")

// Advanced filtering conditions
val eligibleCustomers = table.filter(
  ($"age" >= lit(18)) &&
  ($"creditScore" > lit(650)) &&
  (($"income" > lit(50000)) || ($"assets" > lit(100000))) &&
  !($"blacklisted" === lit(true))
)

// Complex mathematical expressions
val distanceFormula = (($"x2" - $"x1") * ($"x2" - $"x1")) + 
                     (($"y2" - $"y1") * ($"y2" - $"y1"))

val normalizedScore = ($"rawScore" - $"minScore") / 
                     ($"maxScore" - $"minScore")

// Statistical calculations
val zScore = ($"value" - $"mean") / $"standardDeviation"

// String operations (when combined with built-in functions)
val fullName = concat($"firstName", lit(" "), $"lastName")
val emailDomain = substring($"email", position($"email", lit("@")) + lit(1))

// Date arithmetic
val daysSinceOrder = currentDate() - $"orderDate"
val futureDate = $"startDate" + interval(lit(30), DAY)

val intResult: Expression = $"intField" + lit(5)        // Type: INT
val doubleResult: Expression = $"price" * lit(1.1)     // Type: DOUBLE  
val stringResult: Expression = $"name" + lit(" Jr.")   // Type: STRING
val boolResult: Expression = $"age" > lit(18)          // Type: BOOLEAN

// Null-safe operations  
val safeAdd = $"nullableField" + lit(10)         // null + 10 = null
val safeCompare = $"nullableField" > lit(5)      // null > 5 = null
val safeLogical = $"boolField" && lit(true)      // null && true = null

val intDouble = $"intField" + lit(3.14)          // int + double = double
val intLong = $"intField" + lit(100L)           // int + long = long
val numericString = $"price" + lit(" USD")       // double + string = string

// Break complex expressions into readable parts
val isEligibleAge = $"age" >= lit(18)
val hasGoodCredit = $"creditScore" > lit(650)  
val hasIncome = $"income" > lit(30000)

val eligibleCustomer = isEligibleAge && hasGoodCredit && hasIncome

// Explicit null checks when needed
val safeCalculation = $"value".isNotNull ? 
                     ($"value" * lit(1.1)) : 
                     lit(0.0)

// Use appropriate literal types
val intOp = $"intField" + lit(5)           // lit(5) creates INT literal
val doubleOp = $"doubleField" + lit(5.0)   // lit(5.0) creates DOUBLE literal
val stringOp = $"textField" + lit("suffix") // lit("suffix") creates STRING literal