#
# 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.
#

import sys
import json
import warnings
from typing import (
    cast,
    overload,
    Any,
    Callable,
    Iterable,
    List,
    Optional,
    Tuple,
    TYPE_CHECKING,
    Union,
)

from py4j.java_gateway import JavaObject  # type: ignore[import]

from pyspark import copy_func
from pyspark.context import SparkContext
from pyspark.sql.types import DataType, StructField, StructType, IntegerType, StringType

if TYPE_CHECKING:
    from pyspark.sql._typing import ColumnOrName, LiteralType, DecimalLiteral, DateTimeLiteral
    from pyspark.sql.window import WindowSpec

__all__ = ["Column"]


def _create_column_from_literal(literal: Union["LiteralType", "DecimalLiteral"]) -> "Column":
    sc = SparkContext._active_spark_context  # type: ignore[attr-defined]
    return sc._jvm.functions.lit(literal)


def _create_column_from_name(name: str) -> "Column":
    sc = SparkContext._active_spark_context  # type: ignore[attr-defined]
    return sc._jvm.functions.col(name)


def _to_java_column(col: "ColumnOrName") -> JavaObject:
    if isinstance(col, Column):
        jcol = col._jc
    elif isinstance(col, str):
        jcol = _create_column_from_name(col)
    else:
        raise TypeError(
            "Invalid argument, not a string or column: "
            "{0} of type {1}. "
            "For column literals, use 'lit', 'array', 'struct' or 'create_map' "
            "function.".format(col, type(col))
        )
    return jcol


def _to_seq(
    sc: SparkContext,
    cols: Iterable["ColumnOrName"],
    converter: Optional[Callable[["ColumnOrName"], JavaObject]] = None,
) -> JavaObject:
    """
    Convert a list of Column (or names) into a JVM Seq of Column.

    An optional `converter` could be used to convert items in `cols`
    into JVM Column objects.
    """
    if converter: