You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/09/01 03:45:31 UTC
(spark-connect-rust) branch master updated: [SPARK-59140] Close PySpark 4.2.0 public-API parity gaps in the Rust drop-in
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark-connect-rust.git
The following commit(s) were added to refs/heads/master by this push:
new e187281 [SPARK-59140] Close PySpark 4.2.0 public-API parity gaps in the Rust drop-in
e187281 is described below
commit e187281c51639a8cd3960586dc2132d58ae4dd5c
Author: Hyukjin Kwon <hy...@databricks.com>
AuthorDate: Tue Sep 1 12:45:22 2026 +0900
[SPARK-59140] Close PySpark 4.2.0 public-API parity gaps in the Rust drop-in
### What changes were proposed in this pull request?
Brings the `pyspark-client-rust` public API to **full** keyword-argument and signature parity with **PySpark 4.2.0** (`pyspark.sql.connect.*`). Verified by an independent signature harness that diffs `inspect.signature` for every public class/method/function against the reference `v4.2.0` tag:
| | Before | After |
|---|---|---|
| Function keyword-arg mismatches (`*args`) | **513** | **0** |
| Class-method param mismatches | **70** | **0** |
**GAP 1 — functions.** `scripts/gen_fn_dispatch.py` now emits explicit `def`s carrying each function's real 4.2.0 parameter names. Optional params default to an `_UNSET` sentinel and are forwarded only when supplied, so the historical positional dispatch is preserved byte-for-byte (golden tests unchanged) while keyword calls now work: `F.first(col, ignorenulls=True)`, `F.round(col, scale=2)`, `F.split(str, pattern=..., limit=...)`, `F.from_json(col, schema=...)`, etc. The hand-written [...]
**GAP 2 — methods.** camelCase param renames and re-added/wired dropped params across `dataframe.rs`, `column.rs`, `group.rs`, `catalog.rs`, `readwriter.rs`, `streaming.rs`, `session.rs`, `stat.rs`: `sample(withReplacement)`, `sort`/`orderBy(ascending)`, `drop`/`toDF`/`selectExpr`, `melt`/`unpivot(variableColumnName, valueColumnName)`, `persist(storageLevel)`, all `Catalog` methods, `tableExists`/`functionExists(dbName)`, the reader/writer/streaming option lists, `save`/`saveAsTable(f [...]
A handful of params are accepted for signature parity but not yet behaviorally wired because the connect **core** lacks the field (clearly commented in-code): `mapInPandas/mapInArrow(profile)`, `localCheckpoint(storageLevel)`, `printSchema(level)`, `createDataFrame(verifySchema)`, `trigger(realTime)` (raises `NotImplementedError`), and multi-path `read.parquet(*paths)`. These are follow-ups requiring core/proto changes; the client signatures are now correct so no call raises `TypeError`.
### Why are the changes needed?
The project targets "100% public-API parity with PySpark 4.2.0 / runs unchanged." In practice, keyword-argument calls raised `TypeError` on ~507 functions and dozens of methods, and several parameters were missing — so common PySpark code did not run unchanged.
### Does this PR introduce any user-facing change?
Yes — many more calls now work as they do in reference PySpark (keyword arguments across functions/methods, and previously-missing parameters). No behavior change for calls that already worked (positional dispatch is preserved).
### How was this patch tested?
- Signature harness against the `v4.2.0` tag: **0** function mismatches, **0** class-method mismatches.
- All Rust golden/core tests pass (`cargo test -p apache-spark-connect`): **0 failed**.
- New offline unit tests in `python/tests/test_dropin_offline.py` cover the parity paths (keyword dispatch, `col`/`lit`/`expr`/`sha2`/`window`, `lag`/`lead`, the `*args` fallback, `alias(metadata=)`); Python coverage stays **> 90%**.
- `scripts/e2e_parity_check.py` runs **16/16 green** against a real Spark Connect **4.2.0** server, exercising keyword function calls, `sample(withReplacement=)`, `sort(ascending=)`, `createDataFrame(verifySchema=)`, `sql(**kwargs)`, and a `write.save(format=, mode=)` / `read.load(format=)` round-trip.
This pull request and its description were written by Isaac.
Closes #94 from HyukjinKwon/SPARK-XXXXX-close-pyspark-parity-gaps.
Authored-by: Hyukjin Kwon <hy...@databricks.com>
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
crates/pyspark-rs/src/catalog.rs | 89 +-
crates/pyspark-rs/src/column.rs | 41 +-
crates/pyspark-rs/src/dataframe.rs | 324 +++-
crates/pyspark-rs/src/group.rs | 40 +-
crates/pyspark-rs/src/readwriter.rs | 47 +-
crates/pyspark-rs/src/session.rs | 76 +-
crates/pyspark-rs/src/stat.rs | 16 +-
crates/pyspark-rs/src/streaming.rs | 448 ++++-
crates/spark-connect/src/catalog.rs | 11 +
python/pyspark/sql/functions.py | 118 +-
python/pyspark/sql/functions_generated.py | 2923 ++++++++++++++++++++++++-----
python/tests/test_dropin_offline.py | 40 +
scripts/e2e_parity_check.py | 126 ++
scripts/gen_fn_dispatch.py | 117 +-
14 files changed, 3619 insertions(+), 797 deletions(-)
diff --git a/crates/pyspark-rs/src/catalog.rs b/crates/pyspark-rs/src/catalog.rs
index e4736f2..1e1e930 100644
--- a/crates/pyspark-rs/src/catalog.rs
+++ b/crates/pyspark-rs/src/catalog.rs
@@ -225,8 +225,8 @@ impl PyCatalog {
}
/// Set the current catalog.
- fn setCurrentCatalog(&self, catalog_name: &str) -> PyResult<()> {
- self.catalog.set_current_catalog(catalog_name).to_pyerr()
+ fn setCurrentCatalog(&self, catalogName: &str) -> PyResult<()> {
+ self.catalog.set_current_catalog(catalogName).to_pyerr()
}
/// List catalogs. Mirrors `Catalog.listCatalogs(pattern=None) -> List[CatalogMetadata]`.
@@ -247,8 +247,8 @@ impl PyCatalog {
}
/// Set current database.
- fn setCurrentDatabase(&self, db_name: &str) -> PyResult<()> {
- self.catalog.set_current_database(db_name).to_pyerr()
+ fn setCurrentDatabase(&self, dbName: &str) -> PyResult<()> {
+ self.catalog.set_current_database(dbName).to_pyerr()
}
/// List databases. Mirrors `Catalog.listDatabases(pattern=None) -> List[Database]`.
@@ -264,14 +264,14 @@ impl PyCatalog {
}
/// Check if database exists.
- fn databaseExists(&self, db_name: &str) -> PyResult<bool> {
- self.catalog.database_exists(db_name).to_pyerr()
+ fn databaseExists(&self, dbName: &str) -> PyResult<bool> {
+ self.catalog.database_exists(dbName).to_pyerr()
}
/// Get database info. Mirrors `Catalog.getDatabase(dbName) -> Database`.
- fn getDatabase(&self, db_name: &str) -> PyResult<PyDatabase> {
+ fn getDatabase(&self, dbName: &str) -> PyResult<PyDatabase> {
Ok(PyDatabase {
- inner: self.catalog.get_database_typed(db_name).to_pyerr()?,
+ inner: self.catalog.get_database_typed(dbName).to_pyerr()?,
})
}
@@ -288,15 +288,18 @@ impl PyCatalog {
.collect())
}
- /// Check if table exists.
- fn tableExists(&self, table_name: &str) -> PyResult<bool> {
- self.catalog.table_exists(table_name).to_pyerr()
+ /// Check if table exists. Mirrors `Catalog.tableExists(tableName, dbName=None) -> bool`.
+ #[pyo3(signature = (tableName, dbName=None))]
+ fn tableExists(&self, tableName: &str, dbName: Option<&str>) -> PyResult<bool> {
+ self.catalog
+ .table_exists_with_database(tableName, dbName)
+ .to_pyerr()
}
/// Get table info. Mirrors `Catalog.getTable(tableName) -> Table`.
- fn getTable(&self, table_name: &str) -> PyResult<PyTable> {
+ fn getTable(&self, tableName: &str) -> PyResult<PyTable> {
Ok(PyTable {
- inner: self.catalog.get_table_typed(table_name).to_pyerr()?,
+ inner: self.catalog.get_table_typed(tableName).to_pyerr()?,
})
}
@@ -317,15 +320,18 @@ impl PyCatalog {
.collect())
}
- /// Check if function exists.
- fn functionExists(&self, function_name: &str) -> PyResult<bool> {
- self.catalog.function_exists(function_name).to_pyerr()
+ /// Check if function exists. Mirrors `Catalog.functionExists(functionName, dbName=None) -> bool`.
+ #[pyo3(signature = (functionName, dbName=None))]
+ fn functionExists(&self, functionName: &str, dbName: Option<&str>) -> PyResult<bool> {
+ self.catalog
+ .function_exists_with_database(functionName, dbName)
+ .to_pyerr()
}
/// Get function info. Mirrors `Catalog.getFunction(functionName) -> Function`.
- fn getFunction(&self, function_name: &str) -> PyResult<PyFunction> {
+ fn getFunction(&self, functionName: &str) -> PyResult<PyFunction> {
Ok(PyFunction {
- inner: self.catalog.get_function_typed(function_name).to_pyerr()?,
+ inner: self.catalog.get_function_typed(functionName).to_pyerr()?,
})
}
@@ -342,29 +348,46 @@ impl PyCatalog {
.collect())
}
- /// Cache a table.
- fn cacheTable(&self, table_name: &str) -> PyResult<()> {
- self.catalog.cache_table(table_name).to_pyerr()
+ /// Cache a table. Mirrors `Catalog.cacheTable(tableName, storageLevel=None) -> None`.
+ ///
+ /// A supplied pyspark `StorageLevel` is mapped onto the `spark.connect`
+ /// `CacheTable.storage_level` field (the same field layout `DataFrame.persist` uses);
+ /// `None` leaves it unset so the server default applies.
+ #[pyo3(signature = (tableName, storageLevel=None))]
+ fn cacheTable(&self, tableName: &str, storageLevel: Option<&Bound<'_, PyAny>>) -> PyResult<()> {
+ let level = match storageLevel {
+ Some(obj) => Some(spark_connect_proto::StorageLevel {
+ use_disk: obj.getattr("useDisk")?.extract()?,
+ use_memory: obj.getattr("useMemory")?.extract()?,
+ use_off_heap: obj.getattr("useOffHeap")?.extract()?,
+ deserialized: obj.getattr("deserialized")?.extract()?,
+ replication: obj.getattr("replication")?.extract()?,
+ }),
+ None => None,
+ };
+ self.catalog
+ .cache_table_with_storage_level(tableName, level)
+ .to_pyerr()
}
/// Uncache a table.
- fn uncacheTable(&self, table_name: &str) -> PyResult<()> {
- self.catalog.uncache_table(table_name).to_pyerr()
+ fn uncacheTable(&self, tableName: &str) -> PyResult<()> {
+ self.catalog.uncache_table(tableName).to_pyerr()
}
/// Drop temporary view.
- fn dropTempView(&self, view_name: &str) -> PyResult<bool> {
- self.catalog.drop_temp_view(view_name).to_pyerr()
+ fn dropTempView(&self, viewName: &str) -> PyResult<bool> {
+ self.catalog.drop_temp_view(viewName).to_pyerr()
}
/// Drop a global temporary view.
- fn dropGlobalTempView(&self, view_name: &str) -> PyResult<bool> {
- self.catalog.drop_global_temp_view(view_name).to_pyerr()
+ fn dropGlobalTempView(&self, viewName: &str) -> PyResult<bool> {
+ self.catalog.drop_global_temp_view(viewName).to_pyerr()
}
/// Whether a table/view is cached.
- fn isCached(&self, table_name: &str) -> PyResult<bool> {
- self.catalog.is_cached(table_name).to_pyerr()
+ fn isCached(&self, tableName: &str) -> PyResult<bool> {
+ self.catalog.is_cached(tableName).to_pyerr()
}
/// Remove all cached tables from the in-memory cache.
@@ -373,8 +396,8 @@ impl PyCatalog {
}
/// Invalidate and refresh cached metadata for a table.
- fn refreshTable(&self, table_name: &str) -> PyResult<()> {
- self.catalog.refresh_table(table_name).to_pyerr()
+ fn refreshTable(&self, tableName: &str) -> PyResult<()> {
+ self.catalog.refresh_table(tableName).to_pyerr()
}
/// Invalidate and refresh cached data for any table at a path.
@@ -383,8 +406,8 @@ impl PyCatalog {
}
/// Recover all the partitions of a table.
- fn recoverPartitions(&self, table_name: &str) -> PyResult<()> {
- self.catalog.recover_partitions(table_name).to_pyerr()
+ fn recoverPartitions(&self, tableName: &str) -> PyResult<()> {
+ self.catalog.recover_partitions(tableName).to_pyerr()
}
/// Create a table from a data source.
diff --git a/crates/pyspark-rs/src/column.rs b/crates/pyspark-rs/src/column.rs
index bbd1860..48280d9 100644
--- a/crates/pyspark-rs/src/column.rs
+++ b/crates/pyspark-rs/src/column.rs
@@ -1,6 +1,7 @@
//! PyO3 wrapper for spark_connect::column::Column.
use pyo3::prelude::*;
+use pyo3::types::PyDict;
use spark_connect::column::Column;
use crate::functions::to_column;
@@ -50,35 +51,35 @@ fn py_obj_to_column(obj: &Bound<'_, PyAny>) -> PyResult<Column> {
impl PyColumn {
/// Alias the column, optionally attaching column metadata (a dict), mirroring
/// `Column.alias(name, metadata=...)`.
- #[pyo3(signature = (*alias, metadata=None))]
- fn alias(
- &self,
- alias: Vec<String>,
- metadata: Option<std::collections::HashMap<String, String>>,
- ) -> PyColumn {
+ #[pyo3(signature = (*alias, **kwargs))]
+ fn alias(&self, alias: Vec<String>, kwargs: Option<&Bound<'_, PyDict>>) -> PyResult<PyColumn> {
// Mirrors Column.alias(*alias, **kwargs): the common single-name form aliases the
- // column; extra names beyond the first (multi-output aliasing) are not applied here.
+ // column; `metadata` (a dict) may be passed via kwargs. Extra names beyond the first
+ // (multi-output aliasing) are not applied here.
+ let metadata: Option<std::collections::HashMap<String, String>> = match kwargs {
+ Some(kw) => match kw.get_item("metadata")? {
+ Some(v) if !v.is_none() => Some(v.extract()?),
+ _ => None,
+ },
+ None => None,
+ };
let name = match alias.first() {
Some(n) => n.clone(),
- None => return PyColumn::new(self.column.clone()),
+ None => return Ok(PyColumn::new(self.column.clone())),
};
- match metadata {
+ Ok(match metadata {
None => PyColumn::new(self.column.clone().alias(&name)),
Some(m) => {
let md: std::collections::BTreeMap<String, String> = m.into_iter().collect();
PyColumn::new(self.column.clone().alias_with_metadata(&name, md))
}
- }
+ })
}
/// Alias for `alias` (pyspark `Column.name`).
- #[pyo3(signature = (*alias, metadata=None))]
- fn name(
- &self,
- alias: Vec<String>,
- metadata: Option<std::collections::HashMap<String, String>>,
- ) -> PyColumn {
- self.alias(alias, metadata)
+ #[pyo3(signature = (*alias, **kwargs))]
+ fn name(&self, alias: Vec<String>, kwargs: Option<&Bound<'_, PyDict>>) -> PyResult<PyColumn> {
+ self.alias(alias, kwargs)
}
/// Apply a transformation function to this column. Mirrors `Column.transform(f)`,
@@ -495,9 +496,9 @@ impl PyColumn {
let v = to_column(&col.bind(py))?;
Ok(PyColumn::new(self.column.clone().with_field(fieldName, v)))
}
- #[pyo3(name = "dropFields", signature = (*field_names))]
- fn drop_fields(&self, field_names: Vec<String>) -> PyColumn {
- let refs: Vec<&str> = field_names.iter().map(|s| s.as_str()).collect();
+ #[pyo3(name = "dropFields", signature = (*fieldNames))]
+ fn drop_fields(&self, fieldNames: Vec<String>) -> PyColumn {
+ let refs: Vec<&str> = fieldNames.iter().map(|s| s.as_str()).collect();
PyColumn::new(self.column.clone().drop_fields(refs))
}
}
diff --git a/crates/pyspark-rs/src/dataframe.rs b/crates/pyspark-rs/src/dataframe.rs
index b06f4d3..7101aad 100644
--- a/crates/pyspark-rs/src/dataframe.rs
+++ b/crates/pyspark-rs/src/dataframe.rs
@@ -88,6 +88,48 @@ pub(crate) fn to_column_list(
Ok(cols)
}
+/// Build sort-order expressions for sort/orderBy/sortWithinPartitions, honoring an
+/// optional `ascending` kwarg (a bool applied to all columns, or a per-column list of
+/// bools/ints). Mirrors reference pyspark, which wraps each column with asc()/desc()
+/// when `ascending` is given; without it, columns are used as-is.
+fn sort_order_exprs(
+ columns: Vec<spark_connect::column::Column>,
+ kwargs: Option<&Bound<'_, PyDict>>,
+) -> PyResult<Vec<spark_connect::expression::Expression>> {
+ let ascending = match kwargs {
+ Some(kw) => kw.get_item("ascending")?,
+ None => None,
+ };
+ let Some(ascending) = ascending else {
+ return Ok(columns.iter().map(|c| c.expression().clone()).collect());
+ };
+ let flags: Vec<bool> = if let Ok(b) = ascending.extract::<bool>() {
+ vec![b; columns.len()]
+ } else if let Ok(i) = ascending.extract::<i64>() {
+ vec![i != 0; columns.len()]
+ } else if let Ok(list) = ascending.extract::<Vec<bool>>() {
+ list
+ } else if let Ok(list) = ascending.extract::<Vec<i64>>() {
+ list.into_iter().map(|i| i != 0).collect()
+ } else {
+ return Err(PyErr::new::<pyo3::exceptions::PyTypeError, _>(
+ "ascending must be a bool, an int, or a list of bools/ints",
+ ));
+ };
+ if flags.len() != columns.len() {
+ return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
+ "the list length of ascending ({}) must equal the number of sort columns ({})",
+ flags.len(),
+ columns.len()
+ )));
+ }
+ Ok(columns
+ .into_iter()
+ .zip(flags)
+ .map(|(c, asc)| (if asc { c.asc() } else { c.desc() }).expression().clone())
+ .collect())
+}
+
/// Map a pyspark `how` string to a JoinType (shared by join / lateralJoin).
fn parse_join_type(how: &str) -> PyResult<JoinType> {
Ok(match how.to_lowercase().as_str() {
@@ -126,9 +168,9 @@ impl PyDataFrame {
}
/// Select using SQL expressions.
- #[pyo3(signature = (*exprs))]
- fn selectExpr(&self, _py: Python<'_>, exprs: Vec<String>) -> PyDataFrame {
- let columns: Vec<_> = exprs
+ #[pyo3(signature = (*expr))]
+ fn selectExpr(&self, _py: Python<'_>, expr: Vec<String>) -> PyDataFrame {
+ let columns: Vec<_> = expr
.iter()
.map(|e| spark_connect::functions::expr(e))
.collect();
@@ -188,9 +230,9 @@ impl PyDataFrame {
}
/// Drop columns.
- #[pyo3(signature = (*names))]
- fn drop(&self, _py: Python<'_>, names: Vec<String>) -> PyDataFrame {
- let col_refs: Vec<&str> = names.iter().map(|s| s.as_str()).collect();
+ #[pyo3(signature = (*cols))]
+ fn drop(&self, _py: Python<'_>, cols: Vec<String>) -> PyDataFrame {
+ let col_refs: Vec<&str> = cols.iter().map(|s| s.as_str()).collect();
PyDataFrame::new(self.dataframe.drop(col_refs))
}
@@ -218,19 +260,28 @@ impl PyDataFrame {
PyDataFrame::new(self.dataframe.drop_duplicates(col_refs))
}
- /// Sort rows.
- #[pyo3(signature = (*cols))]
- fn sort(&self, _py: Python<'_>, cols: Vec<Bound<'_, PyAny>>) -> PyResult<PyDataFrame> {
- let columns = to_column_list(cols)?;
- let exprs: Vec<_> = columns.iter().map(|c| c.expression().clone()).collect();
+ /// Sort rows. `ascending` (a bool or list of bools, via kwargs) mirrors reference
+ /// pyspark: it wraps each sort column with asc()/desc().
+ #[pyo3(signature = (*cols, **kwargs))]
+ fn sort(
+ &self,
+ _py: Python<'_>,
+ cols: Vec<Bound<'_, PyAny>>,
+ kwargs: Option<Bound<'_, PyDict>>,
+ ) -> PyResult<PyDataFrame> {
+ let exprs = sort_order_exprs(to_column_list(cols)?, kwargs.as_ref())?;
Ok(PyDataFrame::new(self.dataframe.sort(exprs)))
}
/// Alias for sort.
- #[pyo3(signature = (*cols))]
- fn orderBy(&self, _py: Python<'_>, cols: Vec<Bound<'_, PyAny>>) -> PyResult<PyDataFrame> {
- let columns = to_column_list(cols)?;
- let exprs: Vec<_> = columns.iter().map(|c| c.expression().clone()).collect();
+ #[pyo3(signature = (*cols, **kwargs))]
+ fn orderBy(
+ &self,
+ _py: Python<'_>,
+ cols: Vec<Bound<'_, PyAny>>,
+ kwargs: Option<Bound<'_, PyDict>>,
+ ) -> PyResult<PyDataFrame> {
+ let exprs = sort_order_exprs(to_column_list(cols)?, kwargs.as_ref())?;
Ok(PyDataFrame::new(self.dataframe.order_by(exprs)))
}
@@ -438,9 +489,9 @@ impl PyDataFrame {
}
/// Convert to DataFrame with new column names.
- #[pyo3(signature = (*names))]
- fn toDF(&self, _py: Python<'_>, names: Vec<String>) -> PyDataFrame {
- let name_refs: Vec<&str> = names.iter().map(|s| s.as_str()).collect();
+ #[pyo3(signature = (*cols))]
+ fn toDF(&self, _py: Python<'_>, cols: Vec<String>) -> PyDataFrame {
+ let name_refs: Vec<&str> = cols.iter().map(|s| s.as_str()).collect();
PyDataFrame::new(self.dataframe.to_df(name_refs))
}
@@ -472,9 +523,18 @@ impl PyDataFrame {
))
}
- /// Grouping sets aggregation: `groupingSets([[a, b], [a], []])`.
- #[pyo3(name = "groupingSets")]
- fn grouping_sets(&self, groupingSets: Vec<Vec<Bound<'_, PyAny>>>) -> PyResult<PyGroupedData> {
+ /// Grouping sets aggregation: `groupingSets([[a, b], [a], []], "a", "b")`.
+ ///
+ /// `*cols` (the overall grouping columns) is accepted for signature parity with
+ /// reference pyspark; the connect core derives the grouping from the sets, so the
+ /// trailing columns are currently informational.
+ #[pyo3(name = "groupingSets", signature = (groupingSets, *cols))]
+ #[allow(unused_variables)]
+ fn grouping_sets(
+ &self,
+ groupingSets: Vec<Vec<Bound<'_, PyAny>>>,
+ cols: Vec<Bound<'_, PyAny>>,
+ ) -> PyResult<PyGroupedData> {
let mut out: Vec<Vec<spark_connect::column::Column>> =
Vec::with_capacity(groupingSets.len());
for s in groupingSets {
@@ -495,9 +555,8 @@ impl PyDataFrame {
/// `toJSON()` returns an RDD in classic PySpark; Spark Connect has no RDD, so
/// the official Connect client raises NotImplementedError - we match that exactly.
/// (Use `df.select(to_json(struct("*")))` for a JSON-string DataFrame.)
- #[pyo3(name = "toJSON", signature = (use_unicode=true))]
- #[allow(unused_variables)]
- fn to_json(&self, use_unicode: bool) -> PyResult<()> {
+ #[pyo3(name = "toJSON")]
+ fn to_json(&self) -> PyResult<()> {
Err(PyErr::new::<pyo3::exceptions::PyNotImplementedError, _>(
"toJSON() is not supported for Spark Connect (it returns an RDD)",
))
@@ -517,16 +576,14 @@ impl PyDataFrame {
}
/// Sort within each partition.
- #[pyo3(name = "sortWithinPartitions", signature = (*cols))]
+ #[pyo3(name = "sortWithinPartitions", signature = (*cols, **kwargs))]
fn sort_within_partitions(
&self,
_py: Python<'_>,
cols: Vec<Bound<'_, PyAny>>,
+ kwargs: Option<Bound<'_, PyDict>>,
) -> PyResult<PyDataFrame> {
- let exprs: Vec<_> = to_column_list(cols)?
- .iter()
- .map(|c| c.expression().clone())
- .collect();
+ let exprs = sort_order_exprs(to_column_list(cols)?, kwargs.as_ref())?;
Ok(PyDataFrame::new(
self.dataframe.sort_within_partitions(exprs),
))
@@ -763,10 +820,17 @@ impl PyDataFrame {
))
}
- /// Local checkpoint.
- #[pyo3(name = "localCheckpoint", signature = (eager=true))]
+ /// Local checkpoint. `storageLevel` is accepted for signature parity with reference
+ /// pyspark; the connect core's local checkpoint does not yet take a storage level,
+ /// so a supplied value is currently a no-op.
+ #[pyo3(name = "localCheckpoint", signature = (eager=true, storageLevel=None))]
#[allow(unused_variables)]
- fn local_checkpoint(&self, py: Python<'_>, eager: bool) -> PyResult<PyDataFrame> {
+ fn local_checkpoint(
+ &self,
+ py: Python<'_>,
+ eager: bool,
+ storageLevel: Option<Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrame> {
Ok(PyDataFrame::new(
py.detach(|| self.dataframe.local_checkpoint()).to_pyerr()?,
))
@@ -984,53 +1048,63 @@ impl PyDataFrame {
}
/// Unpivot (wide-to-long). `values=None` unpivots all non-id columns.
- #[pyo3(signature = (ids, values=None, var_name="variable", value_name="value"))]
+ #[pyo3(signature = (ids, values=None, variableColumnName="variable", valueColumnName="value"))]
fn unpivot(
&self,
ids: Vec<String>,
values: Option<Vec<String>>,
- var_name: &str,
- value_name: &str,
+ variableColumnName: &str,
+ valueColumnName: &str,
) -> PyDataFrame {
let id_refs: Vec<&str> = ids.iter().map(|s| s.as_str()).collect();
let owned = values;
let val_refs = owned
.as_ref()
.map(|v| v.iter().map(|s| s.as_str()).collect());
- PyDataFrame::new(self.dataframe.melt(id_refs, val_refs, var_name, value_name))
+ PyDataFrame::new(self.dataframe.melt(
+ id_refs,
+ val_refs,
+ variableColumnName,
+ valueColumnName,
+ ))
}
/// Alias for unpivot.
- #[pyo3(signature = (ids, values=None, var_name="variable", value_name="value"))]
+ #[pyo3(signature = (ids, values=None, variableColumnName="variable", valueColumnName="value"))]
fn melt(
&self,
ids: Vec<String>,
values: Option<Vec<String>>,
- var_name: &str,
- value_name: &str,
+ variableColumnName: &str,
+ valueColumnName: &str,
) -> PyDataFrame {
let id_refs: Vec<&str> = ids.iter().map(|s| s.as_str()).collect();
let owned = values;
let val_refs = owned
.as_ref()
.map(|v| v.iter().map(|s| s.as_str()).collect());
- PyDataFrame::new(self.dataframe.melt(id_refs, val_refs, var_name, value_name))
+ PyDataFrame::new(self.dataframe.melt(
+ id_refs,
+ val_refs,
+ variableColumnName,
+ valueColumnName,
+ ))
}
/// Persist the DataFrame. A storage-level argument is accepted for API parity;
/// the default (MEMORY_AND_DISK) is used.
- #[pyo3(signature = (storage_level=None))]
+ #[pyo3(signature = (storageLevel=None))]
fn persist(
&self,
py: Python<'_>,
- storage_level: Option<Bound<'_, PyAny>>,
+ storageLevel: Option<Bound<'_, PyAny>>,
) -> PyResult<PyDataFrame> {
use spark_connect::storage::StorageLevelExt;
// Honor the requested StorageLevel instead of always using MEMORY_AND_DISK.
// A pyspark StorageLevel exposes useDisk/useMemory/useOffHeap/deserialized/
// replication; map them onto the proto. Default (None) is MEMORY_AND_DISK_DESER,
// matching reference DataFrame.persist().
- let level = match storage_level {
+ let level = match storageLevel {
Some(obj) => spark_connect_proto::StorageLevel {
use_disk: obj.getattr("useDisk")?.extract()?,
use_memory: obj.getattr("useMemory")?.extract()?,
@@ -1243,17 +1317,17 @@ impl PyDataFrame {
/// Sample a fraction of rows. Mirrors `DataFrame.sample(withReplacement=None,
/// fraction=None, seed=None)` including the legacy positional form
/// `sample(fraction)` (where the first positional arg is the fraction).
- #[pyo3(signature = (with_replacement=None, fraction=None, seed=None))]
+ #[pyo3(signature = (withReplacement=None, fraction=None, seed=None))]
fn sample(
&self,
- with_replacement: Option<&Bound<'_, PyAny>>,
+ withReplacement: Option<&Bound<'_, PyAny>>,
fraction: Option<f64>,
seed: Option<i64>,
) -> PyResult<PyDataFrame> {
// Resolve the (withReplacement, fraction) overloads exactly like pyspark:
// if the first positional is a float and no fraction was given, it IS the
// fraction (legacy `sample(0.5)`); otherwise it's the bool withReplacement.
- let (replace, frac) = match (with_replacement, fraction) {
+ let (replace, frac) = match (withReplacement, fraction) {
(Some(w), None) => {
if let Ok(f) = w.extract::<f64>() {
(false, f) // legacy sample(fraction)
@@ -1289,8 +1363,12 @@ impl PyDataFrame {
self.dataframe.is_empty().to_pyerr()
}
- /// Print the schema of the DataFrame.
- fn printSchema(&self) -> PyResult<()> {
+ /// Print the schema of the DataFrame. `level` (max nesting depth to print) is
+ /// accepted for signature parity with reference pyspark; the connect core does not
+ /// yet thread a depth into the schema tree-string, so it is currently ignored.
+ #[pyo3(signature = (level=None))]
+ #[allow(unused_variables)]
+ fn printSchema(&self, level: Option<i32>) -> PyResult<()> {
self.dataframe.show(0).to_pyerr()
}
@@ -1374,13 +1452,19 @@ impl PyDataFrame {
}
/// `DataFrame.mapInPandas` — map an iterator of pandas DataFrames.
- #[pyo3(name = "mapInPandas", signature = (func, schema, is_barrier = false))]
+ ///
+ /// `profile` (a ResourceProfile) is accepted for signature parity with reference
+ /// pyspark; the connect core does not yet attach a resource profile id to the
+ /// map-partitions plan, so a supplied profile is currently a no-op.
+ #[pyo3(name = "mapInPandas", signature = (func, schema, barrier = false, profile = None))]
+ #[allow(unused_variables)]
fn map_in_pandas(
&self,
py: Python<'_>,
func: Bound<'_, PyAny>,
schema: Bound<'_, PyAny>,
- is_barrier: bool,
+ barrier: bool,
+ profile: Option<Bound<'_, PyAny>>,
) -> PyResult<PyDataFrame> {
let cols = py.detach(|| self.dataframe.columns()).to_pyerr()?;
let udf = build_map_udf(
@@ -1391,19 +1475,20 @@ impl PyDataFrame {
eval_type::SQL_MAP_PANDAS_ITER_UDF,
&cols,
)?;
- Ok(PyDataFrame::new(
- self.dataframe.map_in_pandas(udf, is_barrier),
- ))
+ Ok(PyDataFrame::new(self.dataframe.map_in_pandas(udf, barrier)))
}
- /// `DataFrame.mapInArrow` — map an iterator of Arrow batches.
- #[pyo3(name = "mapInArrow", signature = (func, schema, is_barrier = false))]
+ /// `DataFrame.mapInArrow` — map an iterator of Arrow batches. `profile` is accepted
+ /// for signature parity (see `mapInPandas`); it is currently a no-op.
+ #[pyo3(name = "mapInArrow", signature = (func, schema, barrier = false, profile = None))]
+ #[allow(unused_variables)]
fn map_in_arrow(
&self,
py: Python<'_>,
func: Bound<'_, PyAny>,
schema: Bound<'_, PyAny>,
- is_barrier: bool,
+ barrier: bool,
+ profile: Option<Bound<'_, PyAny>>,
) -> PyResult<PyDataFrame> {
let cols = py.detach(|| self.dataframe.columns()).to_pyerr()?;
let udf = build_map_udf(
@@ -1414,9 +1499,7 @@ impl PyDataFrame {
eval_type::SQL_MAP_ARROW_ITER_UDF,
&cols,
)?;
- Ok(PyDataFrame::new(
- self.dataframe.map_in_arrow(udf, is_barrier),
- ))
+ Ok(PyDataFrame::new(self.dataframe.map_in_arrow(udf, barrier)))
}
/// `DataFrame.foreach` — run a function per row for side effects.
@@ -1782,11 +1865,47 @@ impl PyDataFrameWriter {
}
}
+/// Apply the shared `format` / `mode` / `partitionBy` / `**options` arguments that
+/// reference `DataFrameWriter.save` and `.saveAsTable` accept inline, mirroring
+/// `.format().mode().partitionBy().options()` chaining.
+fn apply_write_common(
+ mut w: spark_connect::readwriter::DataFrameWriter,
+ format: Option<&str>,
+ mode: Option<&str>,
+ partitionBy: Option<&Bound<'_, PyAny>>,
+ options: Option<&Bound<'_, PyDict>>,
+) -> PyResult<spark_connect::readwriter::DataFrameWriter> {
+ if let Some(f) = format {
+ w = w.format(f);
+ }
+ if let Some(m) = mode {
+ w = w.mode(m);
+ }
+ if let Some(pb) = partitionBy {
+ let cols: Vec<String> = if let Ok(one) = pb.extract::<String>() {
+ vec![one]
+ } else {
+ pb.extract::<Vec<String>>()?
+ };
+ w = w.partition_by(cols);
+ }
+ if let Some(opts) = options {
+ let mut map = std::collections::HashMap::new();
+ for (k, v) in opts.iter() {
+ if let Some(val) = crate::coerce_option_value(&v)? {
+ map.insert(k.str()?.to_string(), val);
+ }
+ }
+ w = w.options(map);
+ }
+ Ok(w)
+}
+
#[pymethods]
impl PyDataFrameWriter {
- fn mode(&mut self, mode: &str) -> PyResult<PyDataFrameWriter> {
+ fn mode(&mut self, saveMode: &str) -> PyResult<PyDataFrameWriter> {
Ok(PyDataFrameWriter {
- inner: Some(self.take()?.mode(mode)),
+ inner: Some(self.take()?.mode(saveMode)),
})
}
@@ -1832,17 +1951,26 @@ impl PyDataFrameWriter {
})
}
- #[pyo3(name = "bucketBy", signature = (num_buckets, *cols))]
- fn bucket_by(&mut self, num_buckets: i32, cols: Vec<String>) -> PyResult<PyDataFrameWriter> {
+ #[pyo3(name = "bucketBy", signature = (numBuckets, col, *cols))]
+ fn bucket_by(
+ &mut self,
+ numBuckets: i32,
+ col: String,
+ cols: Vec<String>,
+ ) -> PyResult<PyDataFrameWriter> {
+ let mut all = vec![col];
+ all.extend(cols);
Ok(PyDataFrameWriter {
- inner: Some(self.take()?.bucket_by(num_buckets, cols)),
+ inner: Some(self.take()?.bucket_by(numBuckets, all)),
})
}
- #[pyo3(name = "sortBy", signature = (*cols))]
- fn sort_by(&mut self, cols: Vec<String>) -> PyResult<PyDataFrameWriter> {
+ #[pyo3(name = "sortBy", signature = (col, *cols))]
+ fn sort_by(&mut self, col: String, cols: Vec<String>) -> PyResult<PyDataFrameWriter> {
+ let mut all = vec![col];
+ all.extend(cols);
Ok(PyDataFrameWriter {
- inner: Some(self.take()?.sort_by(cols)),
+ inner: Some(self.take()?.sort_by(all)),
})
}
@@ -1876,19 +2004,43 @@ impl PyDataFrameWriter {
w.save(None).to_pyerr()
}
- #[pyo3(signature = (path=None))]
- fn save(&mut self, path: Option<String>) -> PyResult<()> {
- self.take()?.save(path.as_deref()).to_pyerr()
+ #[pyo3(signature = (path=None, format=None, mode=None, partitionBy=None, **options))]
+ #[allow(non_snake_case)]
+ fn save(
+ &mut self,
+ path: Option<String>,
+ format: Option<&str>,
+ mode: Option<&str>,
+ partitionBy: Option<&Bound<'_, PyAny>>,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<()> {
+ let w = apply_write_common(self.take()?, format, mode, partitionBy, options)?;
+ w.save(path.as_deref()).to_pyerr()
}
- #[pyo3(name = "saveAsTable")]
- fn save_as_table(&mut self, table_name: &str) -> PyResult<()> {
- self.take()?.save_as_table(table_name).to_pyerr()
+ #[pyo3(name = "saveAsTable", signature = (name, format=None, mode=None, partitionBy=None, **options))]
+ #[allow(non_snake_case)]
+ fn save_as_table(
+ &mut self,
+ name: &str,
+ format: Option<&str>,
+ mode: Option<&str>,
+ partitionBy: Option<&Bound<'_, PyAny>>,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<()> {
+ let w = apply_write_common(self.take()?, format, mode, partitionBy, options)?;
+ w.save_as_table(name).to_pyerr()
}
- #[pyo3(name = "insertInto")]
- fn insert_into(&mut self, table_name: &str) -> PyResult<()> {
- self.take()?.insert_into(table_name).to_pyerr()
+ /// Insert into an existing table. `overwrite` maps to overwrite/append mode,
+ /// mirroring reference `DataFrameWriter.insertInto(tableName, overwrite=None)`.
+ #[pyo3(name = "insertInto", signature = (tableName, overwrite=None))]
+ fn insert_into(&mut self, tableName: &str, overwrite: Option<bool>) -> PyResult<()> {
+ let mut w = self.take()?;
+ if let Some(ow) = overwrite {
+ w = w.mode(if ow { "overwrite" } else { "append" });
+ }
+ w.insert_into(tableName).to_pyerr()
}
/// Write as PARQUET - full pyspark signature; each named option is
@@ -2136,18 +2288,26 @@ impl PyDataFrameWriterV2 {
})
}
- #[pyo3(name = "partitionedBy", signature = (*cols))]
- fn partitioned_by(&mut self, cols: Vec<Bound<'_, PyAny>>) -> PyResult<PyDataFrameWriterV2> {
- let columns = to_column_list(cols)?;
+ #[pyo3(name = "partitionedBy", signature = (col, *cols))]
+ fn partitioned_by(
+ &mut self,
+ col: Bound<'_, PyAny>,
+ cols: Vec<Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrameWriterV2> {
+ let mut all = vec![col];
+ all.extend(cols);
+ let columns = to_column_list(all)?;
Ok(PyDataFrameWriterV2 {
inner: Some(self.take()?.partition_by(columns)),
})
}
- #[pyo3(name = "clusterBy", signature = (*cols))]
- fn cluster_by(&mut self, cols: Vec<String>) -> PyResult<PyDataFrameWriterV2> {
+ #[pyo3(name = "clusterBy", signature = (col, *cols))]
+ fn cluster_by(&mut self, col: String, cols: Vec<String>) -> PyResult<PyDataFrameWriterV2> {
+ let mut all = vec![col];
+ all.extend(cols);
Ok(PyDataFrameWriterV2 {
- inner: Some(self.take()?.cluster_by(cols)),
+ inner: Some(self.take()?.cluster_by(all)),
})
}
diff --git a/crates/pyspark-rs/src/group.rs b/crates/pyspark-rs/src/group.rs
index 8621926..dd0388f 100644
--- a/crates/pyspark-rs/src/group.rs
+++ b/crates/pyspark-rs/src/group.rs
@@ -24,12 +24,12 @@ impl PyGroupedData {
impl PyGroupedData {
/// Aggregate with expressions (accepts Column objects), or the dict form
/// `agg({"col": "aggfunc"})` (PySpark parity), e.g. `agg({"age": "max"})`.
- #[pyo3(signature = (*cols))]
- fn agg(&self, _py: Python<'_>, cols: Vec<Bound<'_, PyAny>>) -> PyResult<PyDataFrame> {
+ #[pyo3(signature = (*exprs))]
+ fn agg(&self, _py: Python<'_>, exprs: Vec<Bound<'_, PyAny>>) -> PyResult<PyDataFrame> {
// Dict form: {column_name: aggregate_function_name} -> func(col(name)) per entry.
- if cols.len() == 1 {
- if let Ok(dict) = cols[0].cast::<pyo3::types::PyDict>() {
- let mut exprs = vec![];
+ if exprs.len() == 1 {
+ if let Ok(dict) = exprs[0].cast::<pyo3::types::PyDict>() {
+ let mut agg_exprs = vec![];
for (k, v) in dict.iter() {
let col_name: String = k.extract()?;
let func_name: String = v.extract()?;
@@ -43,18 +43,18 @@ impl PyGroupedData {
vec![c.expression().clone()],
),
);
- exprs.push(agg_expr);
+ agg_exprs.push(agg_expr);
}
- let df = self.grouped_data.agg(exprs);
+ let df = self.grouped_data.agg(agg_exprs);
return Ok(PyDataFrame::new(df));
}
}
- let mut exprs = vec![];
- for col in cols {
+ let mut agg_exprs = vec![];
+ for col in exprs {
let rust_col = to_column(&col)?;
- exprs.push(rust_col.expression().clone());
+ agg_exprs.push(rust_col.expression().clone());
}
- let df = self.grouped_data.agg(exprs);
+ let df = self.grouped_data.agg(agg_exprs);
Ok(PyDataFrame::new(df))
}
@@ -208,32 +208,32 @@ impl PyGroupedData {
/// `GroupedData.applyInPandasWithState` (stateful streaming).
#[pyo3(name = "applyInPandasWithState")]
- #[pyo3(signature = (func, output_struct_type, state_struct_type, output_mode, timeout_conf))]
+ #[pyo3(signature = (func, outputStructType, stateStructType, outputMode, timeoutConf))]
fn apply_in_pandas_with_state(
&self,
py: Python<'_>,
func: Bound<'_, PyAny>,
- output_struct_type: Bound<'_, PyAny>,
- state_struct_type: Bound<'_, PyAny>,
- output_mode: &str,
- timeout_conf: &str,
+ outputStructType: Bound<'_, PyAny>,
+ stateStructType: Bound<'_, PyAny>,
+ outputMode: &str,
+ timeoutConf: &str,
) -> PyResult<PyDataFrame> {
let cols = py.detach(|| self.grouped_data.input_columns()).to_pyerr()?;
let udf = build_map_udf(
py,
"applyInPandasWithState",
&func,
- &output_struct_type,
+ &outputStructType,
eval_type::SQL_GROUPED_MAP_PANDAS_UDF_WITH_STATE,
&cols,
)?;
- let state_schema = crate::dataframe::resolve_datatype(&state_struct_type)?;
+ let state_schema = crate::dataframe::resolve_datatype(&stateStructType)?;
Ok(PyDataFrame::new(
self.grouped_data.apply_in_pandas_with_state(
udf,
state_schema,
- output_mode,
- timeout_conf,
+ outputMode,
+ timeoutConf,
),
))
}
diff --git a/crates/pyspark-rs/src/readwriter.rs b/crates/pyspark-rs/src/readwriter.rs
index 75177fd..3ef57a5 100644
--- a/crates/pyspark-rs/src/readwriter.rs
+++ b/crates/pyspark-rs/src/readwriter.rs
@@ -104,15 +104,31 @@ impl PyDataFrameReader {
Ok(PyDataFrameReader::new(self.take()?.options(map)))
}
- /// Load data from the (optional) path using the configured format/options.
- #[pyo3(signature = (path=None))]
- fn load(&mut self, path: Option<&str>) -> PyResult<PyDataFrame> {
- Ok(PyDataFrame::new(self.take()?.load(path)))
+ /// Load data from the (optional) path. Mirrors the reference
+ /// `DataFrameReader.load(path=None, format=None, schema=None, **options)`: the
+ /// one-shot form applies format/schema/options here before reading, equivalent to
+ /// chaining `.format(...).schema(...).options(...).load(path)`.
+ #[pyo3(signature = (path=None, format=None, schema=None, **options))]
+ fn load(
+ &mut self,
+ path: Option<&str>,
+ format: Option<&str>,
+ schema: Option<&Bound<'_, PyAny>>,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<PyDataFrame> {
+ let mut r = self.take_with_opts(options)?;
+ if let Some(f) = format {
+ r = r.format(f);
+ }
+ if let Some(s) = schema {
+ r = apply_reader_schema(r, s)?;
+ }
+ Ok(PyDataFrame::new(r.load(path)))
}
/// Read a table by name.
- fn table(&mut self, table_name: &str) -> PyResult<PyDataFrame> {
- Ok(PyDataFrame::new(self.take()?.table(table_name)))
+ fn table(&mut self, tableName: &str) -> PyResult<PyDataFrame> {
+ Ok(PyDataFrame::new(self.take()?.table(tableName)))
}
/// Read the CDC changes of a named table. Mirrors `DataFrameReader.changes`.
@@ -190,14 +206,21 @@ impl PyDataFrameReader {
}
/// Read Parquet file(s).
- #[pyo3(signature = (path, **options))]
+ #[pyo3(signature = (*paths, **options))]
fn parquet(
&mut self,
- path: &str,
+ paths: Vec<String>,
options: Option<&Bound<'_, PyDict>>,
) -> PyResult<PyDataFrame> {
+ // Reference `parquet(*paths)` accepts multiple paths; the connect core reader
+ // currently loads a single path, so we support exactly one here.
+ if paths.len() != 1 {
+ return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
+ "parquet() currently supports exactly one path",
+ ));
+ }
Ok(PyDataFrame::new(
- self.take_with_opts(options)?.parquet(path),
+ self.take_with_opts(options)?.parquet(&paths[0]),
))
}
@@ -309,11 +332,11 @@ impl PyDataFrameReader {
/// Read TEXT file(s) - full pyspark signature; each named option is
/// applied as a read option when provided.
- #[pyo3(signature = (path, wholetext=None, lineSep=None, pathGlobFilter=None, recursiveFileLookup=None, modifiedBefore=None, modifiedAfter=None))]
+ #[pyo3(signature = (paths, wholetext=None, lineSep=None, pathGlobFilter=None, recursiveFileLookup=None, modifiedBefore=None, modifiedAfter=None))]
#[allow(non_snake_case, clippy::too_many_arguments)]
fn text(
&mut self,
- path: &str,
+ paths: &str,
wholetext: Option<&Bound<'_, PyAny>>,
lineSep: Option<&Bound<'_, PyAny>>,
pathGlobFilter: Option<&Bound<'_, PyAny>>,
@@ -328,7 +351,7 @@ impl PyDataFrameReader {
r = set_opt(r, "recursiveFileLookup", recursiveFileLookup)?;
r = set_opt(r, "modifiedBefore", modifiedBefore)?;
r = set_opt(r, "modifiedAfter", modifiedAfter)?;
- Ok(PyDataFrame::new(r.text(path)))
+ Ok(PyDataFrame::new(r.text(paths)))
}
/// Read XML file(s) - full pyspark signature; each named option is
diff --git a/crates/pyspark-rs/src/session.rs b/crates/pyspark-rs/src/session.rs
index 03175f7..c701e94 100644
--- a/crates/pyspark-rs/src/session.rs
+++ b/crates/pyspark-rs/src/session.rs
@@ -251,8 +251,13 @@ impl PySparkSession {
/// Execute a SQL query and return a DataFrame. `args` binds parameters: a list fills
/// positional `?` parameters, a dict fills named (`:name`) parameters. Values are always
/// literals (a string arg is a string literal, never a column reference).
- #[pyo3(signature = (sqlQuery, args=None))]
- fn sql(&self, sqlQuery: &str, args: Option<Bound<'_, PyAny>>) -> PyResult<PyDataFrame> {
+ #[pyo3(signature = (sqlQuery, args=None, **kwargs))]
+ fn sql(
+ &self,
+ sqlQuery: &str,
+ args: Option<Bound<'_, PyAny>>,
+ kwargs: Option<Bound<'_, PyDict>>,
+ ) -> PyResult<PyDataFrame> {
fn to_lit(v: &Bound<'_, PyAny>) -> PyResult<spark_connect::expression::Expression> {
use spark_connect::expression::{Expression, LiteralExpression};
if let Ok(sv) = v.extract::<String>() {
@@ -278,6 +283,13 @@ impl PySparkSession {
));
}
}
+ // Named parameters may also be supplied as **kwargs (reference parity); merge them
+ // into the named-args map alongside any dict passed via `args`.
+ if let Some(kw) = kwargs {
+ for (k, v) in kw.iter() {
+ named.insert(k.extract::<String>()?, to_lit(&v)?);
+ }
+ }
let df = self
.session
.sql_with_args(sqlQuery, pos, named)
@@ -289,15 +301,24 @@ impl PySparkSession {
///
/// If schema is a list of strings, those are used as column names (all types inferred as string).
/// If schema is None, columns are named col0, col1, etc. and all types are string.
- #[pyo3(signature = (data, schema=None))]
+ #[pyo3(signature = (data, schema=None, samplingRatio=None, verifySchema=None))]
fn createDataFrame(
&self,
py: Python<'_>,
data: &Bound<'_, PyAny>,
schema: Option<&Bound<'_, PyAny>>,
+ samplingRatio: Option<f64>,
+ verifySchema: Option<bool>,
) -> PyResult<PyDataFrame> {
use crate::row::PyRow;
+ // `samplingRatio` only affects RDD-based schema inference in reference PySpark and is
+ // unused for local data over Spark Connect (reference ignores it here too).
+ // `verifySchema` toggles client-side row/type verification; this path builds the plan
+ // directly and does not perform that extra verification, so both are accepted for
+ // signature parity. See PARITY_GAP_ANALYSIS.md.
+ let _ = (samplingRatio, verifySchema);
+
// Accept a pandas DataFrame (as pyspark does): convert it to a list of row lists
// with NaN/NaT mapped to None, and take its column names as the default schema.
// pandas-on-Spark always passes an explicit `schema`, so this mainly needs to
@@ -703,11 +724,24 @@ impl PySparkSession {
Ok(PyDataFrame::new(self.session.table(tableName).to_pyerr()?))
}
- #[pyo3(name = "emptyDataFrame")]
- fn empty_data_frame(&self) -> PyResult<PyDataFrame> {
- Ok(PyDataFrame::new(
- self.session.empty_data_frame().to_pyerr()?,
- ))
+ // Reference: emptyDataFrame(schema) -> createDataFrame([], schema). We accept an
+ // optional schema (a permissive superset): with a schema we build a typed empty frame
+ // via createDataFrame; without one we return the core's schema-less empty frame.
+ #[pyo3(name = "emptyDataFrame", signature = (schema=None))]
+ fn empty_data_frame(
+ &self,
+ py: Python<'_>,
+ schema: Option<&Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrame> {
+ match schema {
+ Some(s) => {
+ let empty = PyList::empty(py);
+ self.createDataFrame(py, empty.as_any(), Some(s), None, None)
+ }
+ None => Ok(PyDataFrame::new(
+ self.session.empty_data_frame().to_pyerr()?,
+ )),
+ }
}
#[pyo3(name = "interruptAll")]
@@ -746,10 +780,20 @@ impl PySparkSession {
self.session.clear_tags()
}
- // Mirrors reference `SparkSession.addArtifacts(*path)` - variadic positional paths.
- #[pyo3(name = "addArtifacts", signature = (*paths))]
- fn add_artifacts(&self, py: Python<'_>, paths: Vec<String>) -> PyResult<()> {
- let refs: Vec<&str> = paths.iter().map(|s| s.as_str()).collect();
+ // Mirrors reference `SparkSession.addArtifacts(*path, pyfile=False, archive=False,
+ // file=False)`. The pyfile/archive/file flags are accepted for signature parity;
+ // artifact classification is handled by the Rust transport / server.
+ #[pyo3(name = "addArtifacts", signature = (*path, pyfile=false, archive=false, file=false))]
+ fn add_artifacts(
+ &self,
+ py: Python<'_>,
+ path: Vec<String>,
+ pyfile: bool,
+ archive: bool,
+ file: bool,
+ ) -> PyResult<()> {
+ let _ = (pyfile, archive, file);
+ let refs: Vec<&str> = path.iter().map(|s| s.as_str()).collect();
py.detach(|| self.session.add_artifacts(&refs)).to_pyerr()
}
@@ -774,8 +818,12 @@ impl PySparkSession {
PySparkSession::new(self.session.new_session())
}
- #[pyo3(name = "cloneSession")]
- fn clone_session(&self) -> PySparkSession {
+ // Reference: cloneSession(new_session_id=None). The optional id is accepted for
+ // signature parity; the Rust core assigns the cloned session's id itself and does not
+ // support overriding it. See PARITY_GAP_ANALYSIS.md.
+ #[pyo3(name = "cloneSession", signature = (new_session_id=None))]
+ fn clone_session(&self, new_session_id: Option<String>) -> PySparkSession {
+ let _ = new_session_id;
PySparkSession::new(self.session.clone_session())
}
}
diff --git a/crates/pyspark-rs/src/stat.rs b/crates/pyspark-rs/src/stat.rs
index c545fbf..edb3716 100644
--- a/crates/pyspark-rs/src/stat.rs
+++ b/crates/pyspark-rs/src/stat.rs
@@ -26,7 +26,17 @@ impl PyStatFunctions {
/// the GIL across the call - otherwise a Python thread is blocked for the whole
/// RPC. (`crosstab`/`freq_items`/`approx_quantile` return lazy DataFrames and do
/// no I/O here, so they need no detach.)
- fn corr(&self, py: Python<'_>, col1: &str, col2: &str) -> PyResult<f64> {
+ /// Pearson correlation. `method` is accepted for signature parity with reference
+ /// pyspark, which currently supports only "pearson".
+ #[pyo3(signature = (col1, col2, method=None))]
+ fn corr(&self, py: Python<'_>, col1: &str, col2: &str, method: Option<&str>) -> PyResult<f64> {
+ if let Some(m) = method {
+ if m != "pearson" {
+ return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
+ "Currently only the calculation of the Pearson Correlation coefficient is supported, got {m}"
+ )));
+ }
+ }
py.detach(|| self.stat.corr(col1, col2)).to_pyerr()
}
@@ -77,12 +87,12 @@ impl PyStatFunctions {
py: Python<'_>,
col: &str,
probabilities: Vec<f64>,
- relative_error: f64,
+ relativeError: f64,
) -> PyResult<Vec<f64>> {
use spark_connect::row::Value;
let df = self
.stat
- .approx_quantile(vec![col], probabilities, relative_error);
+ .approx_quantile(vec![col], probabilities, relativeError);
let rows = py.detach(|| df.collect()).to_pyerr()?;
// The server returns a single row, col 0 = array-of-arrays (one inner array of
// quantiles per input column); for a single column we return that inner array.
diff --git a/crates/pyspark-rs/src/streaming.rs b/crates/pyspark-rs/src/streaming.rs
index da40991..09977bc 100644
--- a/crates/pyspark-rs/src/streaming.rs
+++ b/crates/pyspark-rs/src/streaming.rs
@@ -41,6 +41,39 @@ impl StreamingQueryListener for PyListenerAdapter {
}
/// Python wrapper for DataStreamReader.
+/// Apply a named read option to a streaming reader when supplied (mirrors the batch
+/// `set_opt` in readwriter.rs): `None` leaves the option unset (not the string "None");
+/// booleans lowercase to "true"/"false" via `coerce_option_value`.
+fn set_ropt(
+ r: DataStreamReader,
+ name: &str,
+ v: Option<&Bound<'_, PyAny>>,
+) -> PyResult<DataStreamReader> {
+ match v {
+ Some(x) => match crate::coerce_option_value(x)? {
+ Some(sv) => Ok(r.option(name, &sv)),
+ None => Ok(r),
+ },
+ None => Ok(r),
+ }
+}
+
+/// Apply a `schema` argument (a DDL string or a StructType) to a streaming reader,
+/// mirroring the batch `apply_reader_schema`.
+fn apply_stream_schema(r: DataStreamReader, v: &Bound<'_, PyAny>) -> PyResult<DataStreamReader> {
+ let dt = crate::types::py_to_data_type(v)?;
+ Ok(r.schema(dt.simple_string()))
+}
+
+/// Coerce a `partitionBy` argument (a single column name or a list of names) to a
+/// column-name vector, mirroring PySpark's `str | list[str]` acceptance.
+fn extract_partition_cols(v: &Bound<'_, PyAny>) -> PyResult<Vec<String>> {
+ if let Ok(s) = v.extract::<String>() {
+ return Ok(vec![s]);
+ }
+ v.extract::<Vec<String>>()
+}
+
#[pyclass(name = "DataStreamReader", module = "pyspark.sql.streaming.readwriter")]
pub struct PyDataStreamReader {
inner: Option<DataStreamReader>,
@@ -103,10 +136,31 @@ impl PyDataStreamReader {
})
}
- #[pyo3(signature = (path=None))]
- fn load(&mut self, path: Option<&str>) -> PyResult<PyDataFrame> {
- let df = self.take()?.load(path);
- Ok(PyDataFrame::new(df))
+ #[pyo3(signature = (path=None, format=None, schema=None, **options))]
+ fn load(
+ &mut self,
+ path: Option<&str>,
+ format: Option<&str>,
+ schema: Option<&Bound<'_, PyAny>>,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<PyDataFrame> {
+ let mut r = self.take()?;
+ if let Some(fmt) = format {
+ r = r.format(fmt);
+ }
+ if let Some(v) = schema {
+ r = apply_stream_schema(r, v)?;
+ }
+ if let Some(options) = options {
+ let mut opts = HashMap::new();
+ for (k, v) in options.iter() {
+ if let Some(val) = crate::coerce_option_value(&v)? {
+ opts.insert(k.str()?.to_string(), val);
+ }
+ }
+ r = r.options(opts);
+ }
+ Ok(PyDataFrame::new(r.load(path)))
}
/// Set the source name (for checkpoint stability). Mirrors `DataStreamReader.name`,
@@ -131,51 +185,251 @@ impl PyDataStreamReader {
Ok(PyDataFrame::new(self.take()?.changes(tableName)))
}
- /// Load an XML streaming source. Mirrors `DataStreamReader.xml(path, **options)` =
- /// set the options, then `format("xml").load(path)`.
- #[pyo3(signature = (path, **options))]
- #[allow(non_snake_case)]
- fn xml(&mut self, path: &str, options: Option<&Bound<'_, PyDict>>) -> PyResult<PyDataFrame> {
- let mut opts = HashMap::new();
- if let Some(options) = options {
- for (k, v) in options.iter() {
- if let Some(val) = crate::coerce_option_value(&v)? {
- opts.insert(k.str()?.to_string(), val);
- }
- }
+ /// Load an XML streaming source. Mirrors `DataStreamReader.xml(...)`: set each named
+ /// option, then `format("xml").load(path)`.
+ #[pyo3(signature = (path, rowTag=None, schema=None, excludeAttribute=None, attributePrefix=None, valueTag=None, ignoreSurroundingSpaces=None, rowValidationXSDPath=None, ignoreNamespace=None, wildcardColName=None, encoding=None, inferSchema=None, nullValue=None, dateFormat=None, timestampFormat=None, mode=None, columnNameOfCorruptRecord=None, multiLine=None, samplingRatio=None, locale=None))]
+ #[allow(non_snake_case, clippy::too_many_arguments)]
+ fn xml(
+ &mut self,
+ path: &str,
+ rowTag: Option<&Bound<'_, PyAny>>,
+ schema: Option<&Bound<'_, PyAny>>,
+ excludeAttribute: Option<&Bound<'_, PyAny>>,
+ attributePrefix: Option<&Bound<'_, PyAny>>,
+ valueTag: Option<&Bound<'_, PyAny>>,
+ ignoreSurroundingSpaces: Option<&Bound<'_, PyAny>>,
+ rowValidationXSDPath: Option<&Bound<'_, PyAny>>,
+ ignoreNamespace: Option<&Bound<'_, PyAny>>,
+ wildcardColName: Option<&Bound<'_, PyAny>>,
+ encoding: Option<&Bound<'_, PyAny>>,
+ inferSchema: Option<&Bound<'_, PyAny>>,
+ nullValue: Option<&Bound<'_, PyAny>>,
+ dateFormat: Option<&Bound<'_, PyAny>>,
+ timestampFormat: Option<&Bound<'_, PyAny>>,
+ mode: Option<&Bound<'_, PyAny>>,
+ columnNameOfCorruptRecord: Option<&Bound<'_, PyAny>>,
+ multiLine: Option<&Bound<'_, PyAny>>,
+ samplingRatio: Option<&Bound<'_, PyAny>>,
+ locale: Option<&Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrame> {
+ let mut r = self.take()?.format("xml");
+ r = set_ropt(r, "rowTag", rowTag)?;
+ if let Some(v) = schema {
+ r = apply_stream_schema(r, v)?;
}
- let df = self.take()?.format("xml").options(opts).load(Some(path));
- Ok(PyDataFrame::new(df))
- }
-
- fn table(&mut self, table_name: &str) -> PyResult<PyDataFrame> {
- let df = self.take()?.table(table_name);
- Ok(PyDataFrame::new(df))
- }
-
- fn json(&mut self, path: &str) -> PyResult<PyDataFrame> {
- let df = self.take()?.json(path);
- Ok(PyDataFrame::new(df))
- }
-
- fn parquet(&mut self, path: &str) -> PyResult<PyDataFrame> {
- let df = self.take()?.parquet(path);
- Ok(PyDataFrame::new(df))
- }
-
- fn csv(&mut self, path: &str) -> PyResult<PyDataFrame> {
- let df = self.take()?.csv(path);
+ r = set_ropt(r, "excludeAttribute", excludeAttribute)?;
+ r = set_ropt(r, "attributePrefix", attributePrefix)?;
+ r = set_ropt(r, "valueTag", valueTag)?;
+ r = set_ropt(r, "ignoreSurroundingSpaces", ignoreSurroundingSpaces)?;
+ r = set_ropt(r, "rowValidationXSDPath", rowValidationXSDPath)?;
+ r = set_ropt(r, "ignoreNamespace", ignoreNamespace)?;
+ r = set_ropt(r, "wildcardColName", wildcardColName)?;
+ r = set_ropt(r, "encoding", encoding)?;
+ r = set_ropt(r, "inferSchema", inferSchema)?;
+ r = set_ropt(r, "nullValue", nullValue)?;
+ r = set_ropt(r, "dateFormat", dateFormat)?;
+ r = set_ropt(r, "timestampFormat", timestampFormat)?;
+ r = set_ropt(r, "mode", mode)?;
+ r = set_ropt(r, "columnNameOfCorruptRecord", columnNameOfCorruptRecord)?;
+ r = set_ropt(r, "multiLine", multiLine)?;
+ r = set_ropt(r, "samplingRatio", samplingRatio)?;
+ r = set_ropt(r, "locale", locale)?;
+ Ok(PyDataFrame::new(r.load(Some(path))))
+ }
+
+ fn table(&mut self, tableName: &str) -> PyResult<PyDataFrame> {
+ let df = self.take()?.table(tableName);
Ok(PyDataFrame::new(df))
}
- fn orc(&mut self, path: &str) -> PyResult<PyDataFrame> {
- let df = self.take()?.orc(path);
- Ok(PyDataFrame::new(df))
- }
-
- fn text(&mut self, path: &str) -> PyResult<PyDataFrame> {
- let df = self.take()?.text(path);
- Ok(PyDataFrame::new(df))
+ #[pyo3(signature = (path, schema=None, primitivesAsString=None, prefersDecimal=None, allowComments=None, allowUnquotedFieldNames=None, allowSingleQuotes=None, allowNumericLeadingZero=None, allowBackslashEscapingAnyCharacter=None, mode=None, columnNameOfCorruptRecord=None, dateFormat=None, timestampFormat=None, multiLine=None, allowUnquotedControlChars=None, lineSep=None, locale=None, dropFieldIfAllNull=None, encoding=None, pathGlobFilter=None, recursiveFileLookup=None, allowNonNumeri [...]
+ #[allow(non_snake_case, clippy::too_many_arguments)]
+ fn json(
+ &mut self,
+ path: &str,
+ schema: Option<&Bound<'_, PyAny>>,
+ primitivesAsString: Option<&Bound<'_, PyAny>>,
+ prefersDecimal: Option<&Bound<'_, PyAny>>,
+ allowComments: Option<&Bound<'_, PyAny>>,
+ allowUnquotedFieldNames: Option<&Bound<'_, PyAny>>,
+ allowSingleQuotes: Option<&Bound<'_, PyAny>>,
+ allowNumericLeadingZero: Option<&Bound<'_, PyAny>>,
+ allowBackslashEscapingAnyCharacter: Option<&Bound<'_, PyAny>>,
+ mode: Option<&Bound<'_, PyAny>>,
+ columnNameOfCorruptRecord: Option<&Bound<'_, PyAny>>,
+ dateFormat: Option<&Bound<'_, PyAny>>,
+ timestampFormat: Option<&Bound<'_, PyAny>>,
+ multiLine: Option<&Bound<'_, PyAny>>,
+ allowUnquotedControlChars: Option<&Bound<'_, PyAny>>,
+ lineSep: Option<&Bound<'_, PyAny>>,
+ locale: Option<&Bound<'_, PyAny>>,
+ dropFieldIfAllNull: Option<&Bound<'_, PyAny>>,
+ encoding: Option<&Bound<'_, PyAny>>,
+ pathGlobFilter: Option<&Bound<'_, PyAny>>,
+ recursiveFileLookup: Option<&Bound<'_, PyAny>>,
+ allowNonNumericNumbers: Option<&Bound<'_, PyAny>>,
+ useUnsafeRow: Option<&Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrame> {
+ let mut r = self.take()?;
+ if let Some(v) = schema {
+ r = apply_stream_schema(r, v)?;
+ }
+ r = set_ropt(r, "primitivesAsString", primitivesAsString)?;
+ r = set_ropt(r, "prefersDecimal", prefersDecimal)?;
+ r = set_ropt(r, "allowComments", allowComments)?;
+ r = set_ropt(r, "allowUnquotedFieldNames", allowUnquotedFieldNames)?;
+ r = set_ropt(r, "allowSingleQuotes", allowSingleQuotes)?;
+ r = set_ropt(r, "allowNumericLeadingZero", allowNumericLeadingZero)?;
+ r = set_ropt(
+ r,
+ "allowBackslashEscapingAnyCharacter",
+ allowBackslashEscapingAnyCharacter,
+ )?;
+ r = set_ropt(r, "mode", mode)?;
+ r = set_ropt(r, "columnNameOfCorruptRecord", columnNameOfCorruptRecord)?;
+ r = set_ropt(r, "dateFormat", dateFormat)?;
+ r = set_ropt(r, "timestampFormat", timestampFormat)?;
+ r = set_ropt(r, "multiLine", multiLine)?;
+ r = set_ropt(r, "allowUnquotedControlChars", allowUnquotedControlChars)?;
+ r = set_ropt(r, "lineSep", lineSep)?;
+ r = set_ropt(r, "locale", locale)?;
+ r = set_ropt(r, "dropFieldIfAllNull", dropFieldIfAllNull)?;
+ r = set_ropt(r, "encoding", encoding)?;
+ r = set_ropt(r, "pathGlobFilter", pathGlobFilter)?;
+ r = set_ropt(r, "recursiveFileLookup", recursiveFileLookup)?;
+ r = set_ropt(r, "allowNonNumericNumbers", allowNonNumericNumbers)?;
+ r = set_ropt(r, "useUnsafeRow", useUnsafeRow)?;
+ Ok(PyDataFrame::new(r.json(path)))
+ }
+
+ #[pyo3(signature = (path, mergeSchema=None, pathGlobFilter=None, recursiveFileLookup=None, datetimeRebaseMode=None, int96RebaseMode=None))]
+ #[allow(non_snake_case, clippy::too_many_arguments)]
+ fn parquet(
+ &mut self,
+ path: &str,
+ mergeSchema: Option<&Bound<'_, PyAny>>,
+ pathGlobFilter: Option<&Bound<'_, PyAny>>,
+ recursiveFileLookup: Option<&Bound<'_, PyAny>>,
+ datetimeRebaseMode: Option<&Bound<'_, PyAny>>,
+ int96RebaseMode: Option<&Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrame> {
+ let mut r = self.take()?;
+ r = set_ropt(r, "mergeSchema", mergeSchema)?;
+ r = set_ropt(r, "pathGlobFilter", pathGlobFilter)?;
+ r = set_ropt(r, "recursiveFileLookup", recursiveFileLookup)?;
+ r = set_ropt(r, "datetimeRebaseMode", datetimeRebaseMode)?;
+ r = set_ropt(r, "int96RebaseMode", int96RebaseMode)?;
+ Ok(PyDataFrame::new(r.parquet(path)))
+ }
+
+ #[pyo3(signature = (path, schema=None, sep=None, encoding=None, quote=None, escape=None, comment=None, header=None, inferSchema=None, ignoreLeadingWhiteSpace=None, ignoreTrailingWhiteSpace=None, nullValue=None, nanValue=None, positiveInf=None, negativeInf=None, dateFormat=None, timestampFormat=None, maxColumns=None, maxCharsPerColumn=None, maxMalformedLogPerPartition=None, mode=None, columnNameOfCorruptRecord=None, multiLine=None, charToEscapeQuoteEscaping=None, enforceSchema=None, e [...]
+ #[allow(non_snake_case, clippy::too_many_arguments)]
+ fn csv(
+ &mut self,
+ path: &str,
+ schema: Option<&Bound<'_, PyAny>>,
+ sep: Option<&Bound<'_, PyAny>>,
+ encoding: Option<&Bound<'_, PyAny>>,
+ quote: Option<&Bound<'_, PyAny>>,
+ escape: Option<&Bound<'_, PyAny>>,
+ comment: Option<&Bound<'_, PyAny>>,
+ header: Option<&Bound<'_, PyAny>>,
+ inferSchema: Option<&Bound<'_, PyAny>>,
+ ignoreLeadingWhiteSpace: Option<&Bound<'_, PyAny>>,
+ ignoreTrailingWhiteSpace: Option<&Bound<'_, PyAny>>,
+ nullValue: Option<&Bound<'_, PyAny>>,
+ nanValue: Option<&Bound<'_, PyAny>>,
+ positiveInf: Option<&Bound<'_, PyAny>>,
+ negativeInf: Option<&Bound<'_, PyAny>>,
+ dateFormat: Option<&Bound<'_, PyAny>>,
+ timestampFormat: Option<&Bound<'_, PyAny>>,
+ maxColumns: Option<&Bound<'_, PyAny>>,
+ maxCharsPerColumn: Option<&Bound<'_, PyAny>>,
+ maxMalformedLogPerPartition: Option<&Bound<'_, PyAny>>,
+ mode: Option<&Bound<'_, PyAny>>,
+ columnNameOfCorruptRecord: Option<&Bound<'_, PyAny>>,
+ multiLine: Option<&Bound<'_, PyAny>>,
+ charToEscapeQuoteEscaping: Option<&Bound<'_, PyAny>>,
+ enforceSchema: Option<&Bound<'_, PyAny>>,
+ emptyValue: Option<&Bound<'_, PyAny>>,
+ locale: Option<&Bound<'_, PyAny>>,
+ lineSep: Option<&Bound<'_, PyAny>>,
+ pathGlobFilter: Option<&Bound<'_, PyAny>>,
+ recursiveFileLookup: Option<&Bound<'_, PyAny>>,
+ unescapedQuoteHandling: Option<&Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrame> {
+ let mut r = self.take()?;
+ if let Some(v) = schema {
+ r = apply_stream_schema(r, v)?;
+ }
+ r = set_ropt(r, "sep", sep)?;
+ r = set_ropt(r, "encoding", encoding)?;
+ r = set_ropt(r, "quote", quote)?;
+ r = set_ropt(r, "escape", escape)?;
+ r = set_ropt(r, "comment", comment)?;
+ r = set_ropt(r, "header", header)?;
+ r = set_ropt(r, "inferSchema", inferSchema)?;
+ r = set_ropt(r, "ignoreLeadingWhiteSpace", ignoreLeadingWhiteSpace)?;
+ r = set_ropt(r, "ignoreTrailingWhiteSpace", ignoreTrailingWhiteSpace)?;
+ r = set_ropt(r, "nullValue", nullValue)?;
+ r = set_ropt(r, "nanValue", nanValue)?;
+ r = set_ropt(r, "positiveInf", positiveInf)?;
+ r = set_ropt(r, "negativeInf", negativeInf)?;
+ r = set_ropt(r, "dateFormat", dateFormat)?;
+ r = set_ropt(r, "timestampFormat", timestampFormat)?;
+ r = set_ropt(r, "maxColumns", maxColumns)?;
+ r = set_ropt(r, "maxCharsPerColumn", maxCharsPerColumn)?;
+ r = set_ropt(
+ r,
+ "maxMalformedLogPerPartition",
+ maxMalformedLogPerPartition,
+ )?;
+ r = set_ropt(r, "mode", mode)?;
+ r = set_ropt(r, "columnNameOfCorruptRecord", columnNameOfCorruptRecord)?;
+ r = set_ropt(r, "multiLine", multiLine)?;
+ r = set_ropt(r, "charToEscapeQuoteEscaping", charToEscapeQuoteEscaping)?;
+ r = set_ropt(r, "enforceSchema", enforceSchema)?;
+ r = set_ropt(r, "emptyValue", emptyValue)?;
+ r = set_ropt(r, "locale", locale)?;
+ r = set_ropt(r, "lineSep", lineSep)?;
+ r = set_ropt(r, "pathGlobFilter", pathGlobFilter)?;
+ r = set_ropt(r, "recursiveFileLookup", recursiveFileLookup)?;
+ r = set_ropt(r, "unescapedQuoteHandling", unescapedQuoteHandling)?;
+ Ok(PyDataFrame::new(r.csv(path)))
+ }
+
+ #[pyo3(signature = (path, mergeSchema=None, pathGlobFilter=None, recursiveFileLookup=None))]
+ #[allow(non_snake_case)]
+ fn orc(
+ &mut self,
+ path: &str,
+ mergeSchema: Option<&Bound<'_, PyAny>>,
+ pathGlobFilter: Option<&Bound<'_, PyAny>>,
+ recursiveFileLookup: Option<&Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrame> {
+ let mut r = self.take()?;
+ r = set_ropt(r, "mergeSchema", mergeSchema)?;
+ r = set_ropt(r, "pathGlobFilter", pathGlobFilter)?;
+ r = set_ropt(r, "recursiveFileLookup", recursiveFileLookup)?;
+ Ok(PyDataFrame::new(r.orc(path)))
+ }
+
+ #[pyo3(signature = (path, wholetext=None, lineSep=None, pathGlobFilter=None, recursiveFileLookup=None))]
+ #[allow(non_snake_case)]
+ fn text(
+ &mut self,
+ path: &str,
+ wholetext: Option<&Bound<'_, PyAny>>,
+ lineSep: Option<&Bound<'_, PyAny>>,
+ pathGlobFilter: Option<&Bound<'_, PyAny>>,
+ recursiveFileLookup: Option<&Bound<'_, PyAny>>,
+ ) -> PyResult<PyDataFrame> {
+ let mut r = self.take()?;
+ r = set_ropt(r, "wholetext", wholetext)?;
+ r = set_ropt(r, "lineSep", lineSep)?;
+ r = set_ropt(r, "pathGlobFilter", pathGlobFilter)?;
+ r = set_ropt(r, "recursiveFileLookup", recursiveFileLookup)?;
+ Ok(PyDataFrame::new(r.text(path)))
}
}
@@ -241,13 +495,51 @@ impl PyDataStreamWriter {
PyErr::new::<pyo3::exceptions::PyRuntimeError, _>("DataStreamWriter already consumed")
})
}
+
+ /// Apply the one-shot writer options accepted by `start`/`toTable`
+ /// (`format`, `outputMode`, `partitionBy`, `queryName`, `**options`) before
+ /// terminating the stream. Mirrors the reference, where these are keyword args.
+ fn apply_write_opts(
+ &mut self,
+ format: Option<&str>,
+ output_mode: Option<&str>,
+ partition_by: Option<&Bound<'_, PyAny>>,
+ query_name: Option<&str>,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<DataStreamWriter> {
+ let mut w = self.take()?;
+ if let Some(fmt) = format {
+ w = w.format(fmt);
+ }
+ if let Some(m) = output_mode {
+ w = w.output_mode(m);
+ }
+ if let Some(pb) = partition_by {
+ let cols = extract_partition_cols(pb)?;
+ let refs: Vec<&str> = cols.iter().map(|s| s.as_str()).collect();
+ w = w.partition_by(refs);
+ }
+ if let Some(qn) = query_name {
+ w = w.query_name(qn);
+ }
+ if let Some(options) = options {
+ let mut opts = HashMap::new();
+ for (k, v) in options.iter() {
+ if let Some(val) = crate::coerce_option_value(&v)? {
+ opts.insert(k.str()?.to_string(), val);
+ }
+ }
+ w = w.options(opts);
+ }
+ Ok(w)
+ }
}
#[pymethods]
impl PyDataStreamWriter {
- fn outputMode(&mut self, mode: &str) -> PyResult<PyDataStreamWriter> {
+ fn outputMode(&mut self, outputMode: &str) -> PyResult<PyDataStreamWriter> {
Ok(PyDataStreamWriter {
- inner: Some(self.take()?.output_mode(mode)),
+ inner: Some(self.take()?.output_mode(outputMode)),
})
}
@@ -302,33 +594,40 @@ impl PyDataStreamWriter {
})
}
- fn queryName(&mut self, name: &str) -> PyResult<PyDataStreamWriter> {
+ fn queryName(&mut self, queryName: &str) -> PyResult<PyDataStreamWriter> {
Ok(PyDataStreamWriter {
- inner: Some(self.take()?.query_name(name)),
+ inner: Some(self.take()?.query_name(queryName)),
})
}
/// `DataStreamWriter.trigger(...)`: mirrors the reference keyword API — exactly one
/// of `processingTime` / `once` / `availableNow` / `continuous` is given.
- #[pyo3(signature = (processingTime=None, once=None, availableNow=None, continuous=None))]
+ #[pyo3(signature = (*, processingTime=None, once=None, continuous=None, availableNow=None, realTime=None))]
fn trigger(
&mut self,
processingTime: Option<&str>,
once: Option<bool>,
- availableNow: Option<bool>,
continuous: Option<&str>,
+ availableNow: Option<bool>,
+ realTime: Option<&str>,
) -> PyResult<PyDataStreamWriter> {
let trigger = if let Some(interval) = processingTime {
Trigger::ProcessingTime(interval.to_string())
} else if once == Some(true) {
Trigger::Once
- } else if availableNow == Some(true) {
- Trigger::AvailableNow
} else if let Some(interval) = continuous {
Trigger::Continuous(interval.to_string())
+ } else if availableNow == Some(true) {
+ Trigger::AvailableNow
+ } else if realTime.is_some() {
+ // `realTime` is accepted for signature parity with reference pyspark; the
+ // connect core's Trigger enum does not yet model a real-time trigger.
+ return Err(PyErr::new::<pyo3::exceptions::PyNotImplementedError, _>(
+ "the realTime trigger is not yet supported by this client",
+ ));
} else {
return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
- "trigger() requires exactly one of processingTime, once, availableNow, continuous",
+ "trigger() requires exactly one of processingTime, once, continuous, availableNow, realTime",
));
};
Ok(PyDataStreamWriter {
@@ -336,17 +635,36 @@ impl PyDataStreamWriter {
})
}
- #[pyo3(signature = (path=None))]
- fn start(&mut self, path: Option<&str>) -> PyResult<PyStreamingQuery> {
- let writer = self.take()?;
+ #[pyo3(signature = (path=None, format=None, outputMode=None, partitionBy=None, queryName=None, **options))]
+ #[allow(non_snake_case)]
+ fn start(
+ &mut self,
+ path: Option<&str>,
+ format: Option<&str>,
+ outputMode: Option<&str>,
+ partitionBy: Option<&Bound<'_, PyAny>>,
+ queryName: Option<&str>,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<PyStreamingQuery> {
+ let writer = self.apply_write_opts(format, outputMode, partitionBy, queryName, options)?;
// Memory/console/foreach sinks take no path; the core treats "" as unset.
let query = writer.start(path.unwrap_or("")).to_pyerr()?;
Ok(PyStreamingQuery::new(query))
}
- fn toTable(&mut self, table_name: &str) -> PyResult<PyStreamingQuery> {
- let writer = self.take()?;
- let query = writer.to_table(table_name).to_pyerr()?;
+ #[pyo3(signature = (tableName, format=None, outputMode=None, partitionBy=None, queryName=None, **options))]
+ #[allow(non_snake_case)]
+ fn toTable(
+ &mut self,
+ tableName: &str,
+ format: Option<&str>,
+ outputMode: Option<&str>,
+ partitionBy: Option<&Bound<'_, PyAny>>,
+ queryName: Option<&str>,
+ options: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<PyStreamingQuery> {
+ let writer = self.apply_write_opts(format, outputMode, partitionBy, queryName, options)?;
+ let query = writer.to_table(tableName).to_pyerr()?;
Ok(PyStreamingQuery::new(query))
}
@@ -519,8 +837,9 @@ impl PyStreamingQuery {
self.inner.stop().to_pyerr()
}
- fn awaitTermination(&self, timeout_sec: Option<f64>) -> PyResult<Option<bool>> {
- self.inner.await_termination(timeout_sec).to_pyerr()
+ #[pyo3(signature = (timeout=None))]
+ fn awaitTermination(&self, timeout: Option<f64>) -> PyResult<Option<bool>> {
+ self.inner.await_termination(timeout).to_pyerr()
}
#[getter]
@@ -574,8 +893,9 @@ impl PyStreamingQueryManager {
Ok(query.map(PyStreamingQuery::new))
}
- fn awaitAnyTermination(&self, timeout_sec: Option<f64>) -> PyResult<Option<bool>> {
- self.inner.await_any_termination(timeout_sec).to_pyerr()
+ #[pyo3(signature = (timeout=None))]
+ fn awaitAnyTermination(&self, timeout: Option<f64>) -> PyResult<Option<bool>> {
+ self.inner.await_any_termination(timeout).to_pyerr()
}
fn resetTerminated(&self) -> PyResult<()> {
diff --git a/crates/spark-connect/src/catalog.rs b/crates/spark-connect/src/catalog.rs
index 5017881..5de93f8 100644
--- a/crates/spark-connect/src/catalog.rs
+++ b/crates/spark-connect/src/catalog.rs
@@ -626,8 +626,19 @@ impl Catalog {
///
/// Mirrors `pyspark.sql.connect.catalog.Catalog.cacheTable`.
pub fn cache_table(&self, table_name: &str) -> Result<()> {
+ self.cache_table_with_storage_level(table_name, None)
+ }
+
+ /// Cache a table at an optional storage level (`CacheTable.storage_level`, tag 2).
+ /// Mirrors `Catalog.cacheTable(tableName, storageLevel)`.
+ pub fn cache_table_with_storage_level(
+ &self,
+ table_name: &str,
+ storage_level: Option<proto::StorageLevel>,
+ ) -> Result<()> {
let mut cache = proto::CacheTable::default();
cache.table_name = table_name.to_string();
+ cache.storage_level = storage_level;
let mut catalog_msg = proto::Catalog::default();
catalog_msg.cat_type = Some(proto::catalog::CatType::CacheTable(cache));
diff --git a/python/pyspark/sql/functions.py b/python/pyspark/sql/functions.py
index ec00181..47d5644 100644
--- a/python/pyspark/sql/functions.py
+++ b/python/pyspark/sql/functions.py
@@ -228,25 +228,58 @@ _FN_COL_ARGS = {
'when': ((), None),
}
-def _create_wrapper(fname):
- """Create a wrapper for a generated Spark SQL function.
+class _UnsetType:
+ """Sentinel for optional generated-function args that were not supplied.
+
+ Generated wrappers carry PySpark's real parameter names (so keyword calls work),
+ but default every optional parameter to ``_UNSET`` and forward only the args the
+ caller actually passed. This reproduces the historical ``*args`` dispatch exactly
+ (same columns forwarded, in the same order) while accepting keyword arguments.
+ """
+
+ _instance = None
+
+ def __new__(cls):
+ if cls._instance is None:
+ cls._instance = super().__new__(cls)
+ return cls._instance
+
+ def __repr__(self):
+ return "<unset>"
+
+
+_UNSET = _UnsetType()
+
+
+def _dispatch(fname, args):
+ """Convert positional args per the ColumnOrName contract and dispatch to the core.
- PySpark's ColumnOrName contract: a bare ``str`` argument is a COLUMN NAME, so
- column-position ``str`` args are resolved via col() (``_to_col``); literal-position
- args (a format/pattern string, an int count, ...) are passed through unchanged.
- Functions absent from ``_FN_COL_ARGS`` have every positional arg as a column.
+ A bare ``str`` at a column position is a COLUMN NAME (resolved via ``_to_col``);
+ literal-position args (a format/pattern string, an int count, ...) pass through
+ unchanged. Functions absent from ``_FN_COL_ARGS`` have every positional arg as a
+ column. See ``_FN_COL_ARGS``.
"""
override = _FN_COL_ARGS.get(fname)
+ if override is None:
+ conv = [_to_col(a) for a in args]
+ else:
+ cols, vstart = override
+ conv = [
+ _to_col(a) if (i in cols or (vstart is not None and i >= vstart)) else a
+ for i, a in enumerate(args)
+ ]
+ return _wrap(_call_function(fname, *conv))
+
+
+def _create_wrapper(fname):
+ """Fallback wrapper for a generated function with no reference PySpark signature.
+
+ Used only for functions absent from the reference (e.g. Rust-only extras); these
+ keep the historical ``*args`` form. Functions present in the reference get an
+ explicit signature emitted into ``functions_generated.py`` instead.
+ """
def wrapper(*args):
- if override is None:
- conv = [_to_col(a) for a in args]
- else:
- cols, vstart = override
- conv = [
- _to_col(a) if (i in cols or (vstart is not None and i >= vstart)) else a
- for i, a in enumerate(args)
- ]
- return _wrap(_call_function(fname, *conv))
+ return _dispatch(fname, list(args))
wrapper.__doc__ = f"Auto-generated wrapper for {fname}"
wrapper.__name__ = fname
return wrapper
@@ -281,10 +314,22 @@ from pyspark.sql.udtf import (
SelectedColumn,
)
-# Hand-written special functions
-col = _pyfunc_col
-lit = _pyfunc_lit
-expr = _pyfunc_expr
+# Hand-written special functions. Thin wrappers so the public parameter names match
+# the reference PySpark 4.2.0 signatures (keyword-call parity), forwarding to the native
+# bindings whose Rust arg names differ.
+def col(col):
+ """Returns a Column based on the given column name."""
+ return _pyfunc_col(col)
+
+
+def lit(col):
+ """Creates a Column of literal value."""
+ return _pyfunc_lit(col)
+
+
+def expr(str):
+ """Parses the expression string into the column that it represents."""
+ return _pyfunc_expr(str)
sum = _col_wrapper(_pyfunc_sum)
count = _col_wrapper(_pyfunc_count)
avg = _col_wrapper(_pyfunc_avg)
@@ -293,32 +338,32 @@ min = _col_wrapper(_pyfunc_min)
when = _pyfunc_when
# Mixed/special functions - explicit wrappers for non-generic dispatch
-def sha2(col, num_bits):
+def sha2(col, numBits):
"""Returns the hex string result of SHA2 digest of the given data.
Args:
col: column to hash
- num_bits: either 256 or 512
+ numBits: either 256 or 512
"""
- return _pyfunc_sha2(_to_col(col), num_bits)
+ return _pyfunc_sha2(_to_col(col), numBits)
-def window(time_column, window_duration, slide_duration=None, start_time=None):
+def window(timeColumn, windowDuration, slideDuration=None, startTime=None):
"""Buckets rows into one or more time windows specified by the given parameters.
Args:
- time_column: the column containing timestamps
- window_duration: a string specifying the width of the window, e.g. '10 minutes'
- slide_duration: optional, the slide interval. Defaults to ``window_duration``
- (a tumbling window) when only ``start_time`` is given.
- start_time: optional, the offset of the first window, e.g. '15 minutes'.
+ timeColumn: the column containing timestamps
+ windowDuration: a string specifying the width of the window, e.g. '10 minutes'
+ slideDuration: optional, the slide interval. Defaults to ``windowDuration``
+ (a tumbling window) when only ``startTime`` is given.
+ startTime: optional, the offset of the first window, e.g. '15 minutes'.
Defaults to '0 second'.
"""
- if slide_duration is None and start_time is None:
- return _pyfunc_window(_to_col(time_column), window_duration)
- slide = slide_duration if slide_duration is not None else window_duration
- start = start_time if start_time is not None else "0 second"
+ if slideDuration is None and startTime is None:
+ return _pyfunc_window(_to_col(timeColumn), windowDuration)
+ slide = slideDuration if slideDuration is not None else windowDuration
+ start = startTime if startTime is not None else "0 second"
return _pyfunc_window_with_slide_and_start(
- _to_col(time_column), window_duration, slide, start
+ _to_col(timeColumn), windowDuration, slide, start
)
def from_avro(data, json_format_schema):
@@ -410,8 +455,11 @@ spec = importlib.util.spec_from_file_location("functions_generated",
_gen_module = importlib.util.module_from_spec(spec)
# Inject our helper functions into the generated module's namespace before loading
_gen_module._create_wrapper = _create_wrapper
+_gen_module._dispatch = _dispatch
+_gen_module._UNSET = _UNSET
_gen_module._wrap = _wrap
_gen_module._unwrap = _unwrap
+_gen_module._to_col = _to_col
_gen_module._call_function = _call_function
spec.loader.exec_module(_gen_module)
@@ -569,7 +617,9 @@ random = rand
# ``column`` is an alias of ``col`` (matches pyspark.sql.functions).
-column = _pyfunc_col
+def column(col):
+ """Returns a Column based on the given column name (alias of ``col``)."""
+ return _pyfunc_col(col)
def call_udf(udfName, *cols):
diff --git a/python/pyspark/sql/functions_generated.py b/python/pyspark/sql/functions_generated.py
index 5137784..e983a57 100644
--- a/python/pyspark/sql/functions_generated.py
+++ b/python/pyspark/sql/functions_generated.py
@@ -6,506 +6,2342 @@ are defined in functions.py
"""
# Generated wrappers (requires _create_wrapper to be defined)
-abs = _create_wrapper("abs")
-acos = _create_wrapper("acos")
-acosh = _create_wrapper("acosh")
-add_months = _create_wrapper("add_months")
-aes_decrypt = _create_wrapper("aes_decrypt")
-aes_encrypt = _create_wrapper("aes_encrypt")
-any_value = _create_wrapper("any_value")
-approxCountDistinct = _create_wrapper("approxCountDistinct")
-approx_count_distinct = _create_wrapper("approx_count_distinct")
-approx_percentile = _create_wrapper("approx_percentile")
-array = _create_wrapper("array")
-array_agg = _create_wrapper("array_agg")
-array_append = _create_wrapper("array_append")
-array_compact = _create_wrapper("array_compact")
-array_contains = _create_wrapper("array_contains")
-array_distinct = _create_wrapper("array_distinct")
-array_except = _create_wrapper("array_except")
-array_insert = _create_wrapper("array_insert")
-array_intersect = _create_wrapper("array_intersect")
-array_join = _create_wrapper("array_join")
-array_max = _create_wrapper("array_max")
-array_min = _create_wrapper("array_min")
-array_position = _create_wrapper("array_position")
-array_prepend = _create_wrapper("array_prepend")
-array_remove = _create_wrapper("array_remove")
-array_repeat = _create_wrapper("array_repeat")
-array_size = _create_wrapper("array_size")
-array_sort = _create_wrapper("array_sort")
-array_union = _create_wrapper("array_union")
-arrays_overlap = _create_wrapper("arrays_overlap")
-arrays_zip = _create_wrapper("arrays_zip")
-asc = _create_wrapper("asc")
-asc_nulls_first = _create_wrapper("asc_nulls_first")
-asc_nulls_last = _create_wrapper("asc_nulls_last")
-ascii = _create_wrapper("ascii")
-asin = _create_wrapper("asin")
-asinh = _create_wrapper("asinh")
-assert_true = _create_wrapper("assert_true")
-atan = _create_wrapper("atan")
-atan2 = _create_wrapper("atan2")
-atanh = _create_wrapper("atanh")
-base64 = _create_wrapper("base64")
-bin = _create_wrapper("bin")
-bit_and = _create_wrapper("bit_and")
-bit_count = _create_wrapper("bit_count")
-bit_get = _create_wrapper("bit_get")
-bit_length = _create_wrapper("bit_length")
-bit_or = _create_wrapper("bit_or")
-bit_xor = _create_wrapper("bit_xor")
-bitmap_and_agg = _create_wrapper("bitmap_and_agg")
-bitmap_bit_position = _create_wrapper("bitmap_bit_position")
-bitmap_bucket_number = _create_wrapper("bitmap_bucket_number")
-bitmap_construct_agg = _create_wrapper("bitmap_construct_agg")
-bitmap_count = _create_wrapper("bitmap_count")
-bitmap_or_agg = _create_wrapper("bitmap_or_agg")
-bitwiseNOT = _create_wrapper("bitwiseNOT")
-bitwise_not = _create_wrapper("bitwise_not")
-bool_and = _create_wrapper("bool_and")
-bool_or = _create_wrapper("bool_or")
-bround = _create_wrapper("bround")
-btrim = _create_wrapper("btrim")
-bucket = _create_wrapper("bucket")
-cardinality = _create_wrapper("cardinality")
-cbrt = _create_wrapper("cbrt")
-ceil = _create_wrapper("ceil")
-ceiling = _create_wrapper("ceiling")
-char = _create_wrapper("char")
-char_length = _create_wrapper("char_length")
-character_length = _create_wrapper("character_length")
-chr = _create_wrapper("chr")
-coalesce = _create_wrapper("coalesce")
-collate = _create_wrapper("collate")
-collation = _create_wrapper("collation")
-collect_list = _create_wrapper("collect_list")
-collect_set = _create_wrapper("collect_set")
-concat = _create_wrapper("concat")
-concat_ws = _create_wrapper("concat_ws")
-contains = _create_wrapper("contains")
-conv = _create_wrapper("conv")
-convert_timezone = _create_wrapper("convert_timezone")
-corr = _create_wrapper("corr")
-cos = _create_wrapper("cos")
-cosh = _create_wrapper("cosh")
-cot = _create_wrapper("cot")
-countDistinct = _create_wrapper("countDistinct")
-count_distinct = _create_wrapper("count_distinct")
-count_if = _create_wrapper("count_if")
-count_min_sketch = _create_wrapper("count_min_sketch")
+def abs(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("abs", _a)
+def acos(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("acos", _a)
+def acosh(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("acosh", _a)
+def add_months(start, months):
+ _a = []
+ _a.append(start)
+ _a.append(months)
+ return _dispatch("add_months", _a)
+def aes_decrypt(input, key, mode=_UNSET, padding=_UNSET, aad=_UNSET):
+ _a = []
+ _a.append(input)
+ _a.append(key)
+ if mode is not _UNSET: _a.append(mode)
+ if padding is not _UNSET: _a.append(padding)
+ if aad is not _UNSET: _a.append(aad)
+ return _dispatch("aes_decrypt", _a)
+def aes_encrypt(input, key, mode=_UNSET, padding=_UNSET, iv=_UNSET, aad=_UNSET):
+ _a = []
+ _a.append(input)
+ _a.append(key)
+ if mode is not _UNSET: _a.append(mode)
+ if padding is not _UNSET: _a.append(padding)
+ if iv is not _UNSET: _a.append(iv)
+ if aad is not _UNSET: _a.append(aad)
+ return _dispatch("aes_encrypt", _a)
+def any_value(col, ignoreNulls=_UNSET):
+ _a = []
+ _a.append(col)
+ if ignoreNulls is not _UNSET: _a.append(ignoreNulls)
+ return _dispatch("any_value", _a)
+def approxCountDistinct(col, rsd=_UNSET):
+ _a = []
+ _a.append(col)
+ if rsd is not _UNSET: _a.append(rsd)
+ return _dispatch("approxCountDistinct", _a)
+def approx_count_distinct(col, rsd=_UNSET):
+ _a = []
+ _a.append(col)
+ if rsd is not _UNSET: _a.append(rsd)
+ return _dispatch("approx_count_distinct", _a)
+def approx_percentile(col, percentage, accuracy=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(percentage)
+ if accuracy is not _UNSET: _a.append(accuracy)
+ return _dispatch("approx_percentile", _a)
+def array(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("array", _a)
+def array_agg(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("array_agg", _a)
+def array_append(col, value):
+ _a = []
+ _a.append(col)
+ _a.append(value)
+ return _dispatch("array_append", _a)
+def array_compact(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("array_compact", _a)
+def array_contains(col, value):
+ _a = []
+ _a.append(col)
+ _a.append(value)
+ return _dispatch("array_contains", _a)
+def array_distinct(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("array_distinct", _a)
+def array_except(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("array_except", _a)
+def array_insert(arr, pos, value):
+ _a = []
+ _a.append(arr)
+ _a.append(pos)
+ _a.append(value)
+ return _dispatch("array_insert", _a)
+def array_intersect(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("array_intersect", _a)
+def array_join(col, delimiter, null_replacement=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(delimiter)
+ if null_replacement is not _UNSET: _a.append(null_replacement)
+ return _dispatch("array_join", _a)
+def array_max(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("array_max", _a)
+def array_min(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("array_min", _a)
+def array_position(col, value):
+ _a = []
+ _a.append(col)
+ _a.append(value)
+ return _dispatch("array_position", _a)
+def array_prepend(col, value):
+ _a = []
+ _a.append(col)
+ _a.append(value)
+ return _dispatch("array_prepend", _a)
+def array_remove(col, element):
+ _a = []
+ _a.append(col)
+ _a.append(element)
+ return _dispatch("array_remove", _a)
+def array_repeat(col, count):
+ _a = []
+ _a.append(col)
+ _a.append(count)
+ return _dispatch("array_repeat", _a)
+def array_size(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("array_size", _a)
+def array_sort(col, comparator=_UNSET):
+ _a = []
+ _a.append(col)
+ if comparator is not _UNSET: _a.append(comparator)
+ return _dispatch("array_sort", _a)
+def array_union(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("array_union", _a)
+def arrays_overlap(a1, a2):
+ _a = []
+ _a.append(a1)
+ _a.append(a2)
+ return _dispatch("arrays_overlap", _a)
+def arrays_zip(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("arrays_zip", _a)
+def asc(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("asc", _a)
+def asc_nulls_first(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("asc_nulls_first", _a)
+def asc_nulls_last(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("asc_nulls_last", _a)
+def ascii(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("ascii", _a)
+def asin(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("asin", _a)
+def asinh(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("asinh", _a)
+def assert_true(col, errMsg=_UNSET):
+ _a = []
+ _a.append(col)
+ if errMsg is not _UNSET: _a.append(errMsg)
+ return _dispatch("assert_true", _a)
+def atan(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("atan", _a)
+def atan2(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("atan2", _a)
+def atanh(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("atanh", _a)
+def base64(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("base64", _a)
+def bin(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bin", _a)
+def bit_and(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bit_and", _a)
+def bit_count(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bit_count", _a)
+def bit_get(col, pos):
+ _a = []
+ _a.append(col)
+ _a.append(pos)
+ return _dispatch("bit_get", _a)
+def bit_length(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bit_length", _a)
+def bit_or(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bit_or", _a)
+def bit_xor(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bit_xor", _a)
+def bitmap_and_agg(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bitmap_and_agg", _a)
+def bitmap_bit_position(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bitmap_bit_position", _a)
+def bitmap_bucket_number(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bitmap_bucket_number", _a)
+def bitmap_construct_agg(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bitmap_construct_agg", _a)
+def bitmap_count(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bitmap_count", _a)
+def bitmap_or_agg(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bitmap_or_agg", _a)
+def bitwiseNOT(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bitwiseNOT", _a)
+def bitwise_not(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bitwise_not", _a)
+def bool_and(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bool_and", _a)
+def bool_or(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("bool_or", _a)
+def bround(col, scale=_UNSET):
+ _a = []
+ _a.append(col)
+ if scale is not _UNSET: _a.append(scale)
+ return _dispatch("bround", _a)
+def btrim(str, trim=_UNSET):
+ _a = []
+ _a.append(str)
+ if trim is not _UNSET: _a.append(trim)
+ return _dispatch("btrim", _a)
+def bucket(numBuckets, col):
+ _a = []
+ _a.append(numBuckets)
+ _a.append(col)
+ return _dispatch("bucket", _a)
+def cardinality(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("cardinality", _a)
+def cbrt(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("cbrt", _a)
+def ceil(col, scale=_UNSET):
+ _a = []
+ _a.append(col)
+ if scale is not _UNSET: _a.append(scale)
+ return _dispatch("ceil", _a)
+def ceiling(col, scale=_UNSET):
+ _a = []
+ _a.append(col)
+ if scale is not _UNSET: _a.append(scale)
+ return _dispatch("ceiling", _a)
+def char(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("char", _a)
+def char_length(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("char_length", _a)
+def character_length(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("character_length", _a)
+def chr(n):
+ _a = []
+ _a.append(n)
+ return _dispatch("chr", _a)
+def coalesce(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("coalesce", _a)
+def collate(col, collation):
+ _a = []
+ _a.append(col)
+ _a.append(collation)
+ return _dispatch("collate", _a)
+def collation(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("collation", _a)
+def collect_list(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("collect_list", _a)
+def collect_set(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("collect_set", _a)
+def concat(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("concat", _a)
+def concat_ws(sep, *cols):
+ _a = []
+ _a.append(sep)
+ _a.extend(cols)
+ return _dispatch("concat_ws", _a)
+def contains(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("contains", _a)
+def conv(col, fromBase, toBase):
+ _a = []
+ _a.append(col)
+ _a.append(fromBase)
+ _a.append(toBase)
+ return _dispatch("conv", _a)
+def convert_timezone(sourceTz, targetTz, sourceTs):
+ _a = []
+ _a.append(sourceTz)
+ _a.append(targetTz)
+ _a.append(sourceTs)
+ return _dispatch("convert_timezone", _a)
+def corr(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("corr", _a)
+def cos(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("cos", _a)
+def cosh(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("cosh", _a)
+def cot(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("cot", _a)
+def countDistinct(col, *cols):
+ _a = []
+ _a.append(col)
+ _a.extend(cols)
+ return _dispatch("countDistinct", _a)
+def count_distinct(col, *cols):
+ _a = []
+ _a.append(col)
+ _a.extend(cols)
+ return _dispatch("count_distinct", _a)
+def count_if(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("count_if", _a)
+def count_min_sketch(col, eps, confidence, seed=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(eps)
+ _a.append(confidence)
+ if seed is not _UNSET: _a.append(seed)
+ return _dispatch("count_min_sketch", _a)
counter_diff = _create_wrapper("counter_diff")
-covar_pop = _create_wrapper("covar_pop")
-covar_samp = _create_wrapper("covar_samp")
-crc32 = _create_wrapper("crc32")
-create_map = _create_wrapper("create_map")
-csc = _create_wrapper("csc")
-curdate = _create_wrapper("curdate")
-current_catalog = _create_wrapper("current_catalog")
-current_database = _create_wrapper("current_database")
-current_date = _create_wrapper("current_date")
-current_path = _create_wrapper("current_path")
-current_schema = _create_wrapper("current_schema")
-current_time = _create_wrapper("current_time")
-current_timestamp = _create_wrapper("current_timestamp")
-current_timezone = _create_wrapper("current_timezone")
-current_user = _create_wrapper("current_user")
-date_add = _create_wrapper("date_add")
-date_diff = _create_wrapper("date_diff")
-date_format = _create_wrapper("date_format")
-date_from_unix_date = _create_wrapper("date_from_unix_date")
-date_part = _create_wrapper("date_part")
-date_sub = _create_wrapper("date_sub")
-date_trunc = _create_wrapper("date_trunc")
-dateadd = _create_wrapper("dateadd")
-datediff = _create_wrapper("datediff")
-datepart = _create_wrapper("datepart")
-day = _create_wrapper("day")
-dayname = _create_wrapper("dayname")
-dayofmonth = _create_wrapper("dayofmonth")
-dayofweek = _create_wrapper("dayofweek")
-dayofyear = _create_wrapper("dayofyear")
-days = _create_wrapper("days")
-decode = _create_wrapper("decode")
-degrees = _create_wrapper("degrees")
-dense_rank = _create_wrapper("dense_rank")
-desc = _create_wrapper("desc")
-desc_nulls_first = _create_wrapper("desc_nulls_first")
-desc_nulls_last = _create_wrapper("desc_nulls_last")
-e = _create_wrapper("e")
-element_at = _create_wrapper("element_at")
-elt = _create_wrapper("elt")
-encode = _create_wrapper("encode")
-endswith = _create_wrapper("endswith")
-equal_null = _create_wrapper("equal_null")
-every = _create_wrapper("every")
-exp = _create_wrapper("exp")
-explode = _create_wrapper("explode")
-explode_outer = _create_wrapper("explode_outer")
-expm1 = _create_wrapper("expm1")
-extract = _create_wrapper("extract")
-factorial = _create_wrapper("factorial")
-find_in_set = _create_wrapper("find_in_set")
-first = _create_wrapper("first")
-first_value = _create_wrapper("first_value")
-flatten = _create_wrapper("flatten")
-floor = _create_wrapper("floor")
-format_number = _create_wrapper("format_number")
-format_string = _create_wrapper("format_string")
-from_csv = _create_wrapper("from_csv")
-from_json = _create_wrapper("from_json")
-from_unixtime = _create_wrapper("from_unixtime")
-from_utc_timestamp = _create_wrapper("from_utc_timestamp")
-from_xml = _create_wrapper("from_xml")
-get = _create_wrapper("get")
-get_json_object = _create_wrapper("get_json_object")
-getbit = _create_wrapper("getbit")
-greatest = _create_wrapper("greatest")
-grouping = _create_wrapper("grouping")
-grouping_id = _create_wrapper("grouping_id")
-hash = _create_wrapper("hash")
-hex = _create_wrapper("hex")
-histogram_numeric = _create_wrapper("histogram_numeric")
-hll_sketch_agg = _create_wrapper("hll_sketch_agg")
-hll_sketch_estimate = _create_wrapper("hll_sketch_estimate")
-hll_union = _create_wrapper("hll_union")
-hll_union_agg = _create_wrapper("hll_union_agg")
+def covar_pop(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("covar_pop", _a)
+def covar_samp(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("covar_samp", _a)
+def crc32(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("crc32", _a)
+def create_map(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("create_map", _a)
+def csc(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("csc", _a)
+def curdate():
+ _a = []
+ return _dispatch("curdate", _a)
+def current_catalog():
+ _a = []
+ return _dispatch("current_catalog", _a)
+def current_database():
+ _a = []
+ return _dispatch("current_database", _a)
+def current_date():
+ _a = []
+ return _dispatch("current_date", _a)
+def current_path():
+ _a = []
+ return _dispatch("current_path", _a)
+def current_schema():
+ _a = []
+ return _dispatch("current_schema", _a)
+def current_time(precision=_UNSET):
+ _a = []
+ if precision is not _UNSET: _a.append(precision)
+ return _dispatch("current_time", _a)
+def current_timestamp():
+ _a = []
+ return _dispatch("current_timestamp", _a)
+def current_timezone():
+ _a = []
+ return _dispatch("current_timezone", _a)
+def current_user():
+ _a = []
+ return _dispatch("current_user", _a)
+def date_add(start, days):
+ _a = []
+ _a.append(start)
+ _a.append(days)
+ return _dispatch("date_add", _a)
+def date_diff(end, start):
+ _a = []
+ _a.append(end)
+ _a.append(start)
+ return _dispatch("date_diff", _a)
+def date_format(date, format):
+ _a = []
+ _a.append(date)
+ _a.append(format)
+ return _dispatch("date_format", _a)
+def date_from_unix_date(days):
+ _a = []
+ _a.append(days)
+ return _dispatch("date_from_unix_date", _a)
+def date_part(field, source):
+ _a = []
+ _a.append(field)
+ _a.append(source)
+ return _dispatch("date_part", _a)
+def date_sub(start, days):
+ _a = []
+ _a.append(start)
+ _a.append(days)
+ return _dispatch("date_sub", _a)
+def date_trunc(format, timestamp):
+ _a = []
+ _a.append(format)
+ _a.append(timestamp)
+ return _dispatch("date_trunc", _a)
+def dateadd(start, days):
+ _a = []
+ _a.append(start)
+ _a.append(days)
+ return _dispatch("dateadd", _a)
+def datediff(end, start):
+ _a = []
+ _a.append(end)
+ _a.append(start)
+ return _dispatch("datediff", _a)
+def datepart(field, source):
+ _a = []
+ _a.append(field)
+ _a.append(source)
+ return _dispatch("datepart", _a)
+def day(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("day", _a)
+def dayname(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("dayname", _a)
+def dayofmonth(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("dayofmonth", _a)
+def dayofweek(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("dayofweek", _a)
+def dayofyear(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("dayofyear", _a)
+def days(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("days", _a)
+def decode(col, charset):
+ _a = []
+ _a.append(col)
+ _a.append(charset)
+ return _dispatch("decode", _a)
+def degrees(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("degrees", _a)
+def dense_rank():
+ _a = []
+ return _dispatch("dense_rank", _a)
+def desc(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("desc", _a)
+def desc_nulls_first(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("desc_nulls_first", _a)
+def desc_nulls_last(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("desc_nulls_last", _a)
+def e():
+ _a = []
+ return _dispatch("e", _a)
+def element_at(col, extraction):
+ _a = []
+ _a.append(col)
+ _a.append(extraction)
+ return _dispatch("element_at", _a)
+def elt(*inputs):
+ _a = []
+ _a.extend(inputs)
+ return _dispatch("elt", _a)
+def encode(col, charset):
+ _a = []
+ _a.append(col)
+ _a.append(charset)
+ return _dispatch("encode", _a)
+def endswith(str, suffix):
+ _a = []
+ _a.append(str)
+ _a.append(suffix)
+ return _dispatch("endswith", _a)
+def equal_null(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("equal_null", _a)
+def every(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("every", _a)
+def exp(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("exp", _a)
+def explode(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("explode", _a)
+def explode_outer(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("explode_outer", _a)
+def expm1(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("expm1", _a)
+def extract(field, source):
+ _a = []
+ _a.append(field)
+ _a.append(source)
+ return _dispatch("extract", _a)
+def factorial(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("factorial", _a)
+def find_in_set(str, str_array):
+ _a = []
+ _a.append(str)
+ _a.append(str_array)
+ return _dispatch("find_in_set", _a)
+def first(col, ignorenulls=_UNSET):
+ _a = []
+ _a.append(col)
+ if ignorenulls is not _UNSET: _a.append(ignorenulls)
+ return _dispatch("first", _a)
+def first_value(col, ignoreNulls=_UNSET):
+ _a = []
+ _a.append(col)
+ if ignoreNulls is not _UNSET: _a.append(ignoreNulls)
+ return _dispatch("first_value", _a)
+def flatten(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("flatten", _a)
+def floor(col, scale=_UNSET):
+ _a = []
+ _a.append(col)
+ if scale is not _UNSET: _a.append(scale)
+ return _dispatch("floor", _a)
+def format_number(col, d):
+ _a = []
+ _a.append(col)
+ _a.append(d)
+ return _dispatch("format_number", _a)
+def format_string(format, *cols):
+ _a = []
+ _a.append(format)
+ _a.extend(cols)
+ return _dispatch("format_string", _a)
+def from_csv(col, schema, options=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(schema)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("from_csv", _a)
+def from_json(col, schema, options=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(schema)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("from_json", _a)
+def from_unixtime(timestamp, format=_UNSET):
+ _a = []
+ _a.append(timestamp)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("from_unixtime", _a)
+def from_utc_timestamp(timestamp, tz):
+ _a = []
+ _a.append(timestamp)
+ _a.append(tz)
+ return _dispatch("from_utc_timestamp", _a)
+def from_xml(col, schema, options=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(schema)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("from_xml", _a)
+def get(col, index):
+ _a = []
+ _a.append(col)
+ _a.append(index)
+ return _dispatch("get", _a)
+def get_json_object(col, path):
+ _a = []
+ _a.append(col)
+ _a.append(path)
+ return _dispatch("get_json_object", _a)
+def getbit(col, pos):
+ _a = []
+ _a.append(col)
+ _a.append(pos)
+ return _dispatch("getbit", _a)
+def greatest(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("greatest", _a)
+def grouping(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("grouping", _a)
+def grouping_id(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("grouping_id", _a)
+def hash(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("hash", _a)
+def hex(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("hex", _a)
+def histogram_numeric(col, nBins):
+ _a = []
+ _a.append(col)
+ _a.append(nBins)
+ return _dispatch("histogram_numeric", _a)
+def hll_sketch_agg(col, lgConfigK=_UNSET):
+ _a = []
+ _a.append(col)
+ if lgConfigK is not _UNSET: _a.append(lgConfigK)
+ return _dispatch("hll_sketch_agg", _a)
+def hll_sketch_estimate(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("hll_sketch_estimate", _a)
+def hll_union(col1, col2, allowDifferentLgConfigK=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if allowDifferentLgConfigK is not _UNSET: _a.append(allowDifferentLgConfigK)
+ return _dispatch("hll_union", _a)
+def hll_union_agg(col, allowDifferentLgConfigK=_UNSET):
+ _a = []
+ _a.append(col)
+ if allowDifferentLgConfigK is not _UNSET: _a.append(allowDifferentLgConfigK)
+ return _dispatch("hll_union_agg", _a)
hmac = _create_wrapper("hmac")
-hour = _create_wrapper("hour")
-hours = _create_wrapper("hours")
-hypot = _create_wrapper("hypot")
-ifnull = _create_wrapper("ifnull")
-ilike = _create_wrapper("ilike")
-initcap = _create_wrapper("initcap")
-inline = _create_wrapper("inline")
-inline_outer = _create_wrapper("inline_outer")
-input_file_block_length = _create_wrapper("input_file_block_length")
-input_file_block_start = _create_wrapper("input_file_block_start")
-input_file_name = _create_wrapper("input_file_name")
-instr = _create_wrapper("instr")
-is_valid_utf8 = _create_wrapper("is_valid_utf8")
-is_valid_variant = _create_wrapper("is_valid_variant")
-is_variant_null = _create_wrapper("is_variant_null")
-isnan = _create_wrapper("isnan")
-isnotnull = _create_wrapper("isnotnull")
-isnull = _create_wrapper("isnull")
+def hour(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("hour", _a)
+def hours(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("hours", _a)
+def hypot(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("hypot", _a)
+def ifnull(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("ifnull", _a)
+def ilike(str, pattern, escapeChar=_UNSET):
+ _a = []
+ _a.append(str)
+ _a.append(pattern)
+ if escapeChar is not _UNSET: _a.append(escapeChar)
+ return _dispatch("ilike", _a)
+def initcap(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("initcap", _a)
+def inline(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("inline", _a)
+def inline_outer(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("inline_outer", _a)
+def input_file_block_length():
+ _a = []
+ return _dispatch("input_file_block_length", _a)
+def input_file_block_start():
+ _a = []
+ return _dispatch("input_file_block_start", _a)
+def input_file_name():
+ _a = []
+ return _dispatch("input_file_name", _a)
+def instr(str, substr):
+ _a = []
+ _a.append(str)
+ _a.append(substr)
+ return _dispatch("instr", _a)
+def is_valid_utf8(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("is_valid_utf8", _a)
+def is_valid_variant(v):
+ _a = []
+ _a.append(v)
+ return _dispatch("is_valid_variant", _a)
+def is_variant_null(v):
+ _a = []
+ _a.append(v)
+ return _dispatch("is_variant_null", _a)
+def isnan(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("isnan", _a)
+def isnotnull(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("isnotnull", _a)
+def isnull(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("isnull", _a)
jaro_winkler_similarity = _create_wrapper("jaro_winkler_similarity")
-java_method = _create_wrapper("java_method")
-json_array_length = _create_wrapper("json_array_length")
-json_object_keys = _create_wrapper("json_object_keys")
-json_tuple = _create_wrapper("json_tuple")
-kll_merge_agg_bigint = _create_wrapper("kll_merge_agg_bigint")
-kll_merge_agg_double = _create_wrapper("kll_merge_agg_double")
-kll_merge_agg_float = _create_wrapper("kll_merge_agg_float")
-kll_sketch_agg_bigint = _create_wrapper("kll_sketch_agg_bigint")
-kll_sketch_agg_double = _create_wrapper("kll_sketch_agg_double")
-kll_sketch_agg_float = _create_wrapper("kll_sketch_agg_float")
-kll_sketch_get_n_bigint = _create_wrapper("kll_sketch_get_n_bigint")
-kll_sketch_get_n_double = _create_wrapper("kll_sketch_get_n_double")
-kll_sketch_get_n_float = _create_wrapper("kll_sketch_get_n_float")
-kll_sketch_get_quantile_bigint = _create_wrapper("kll_sketch_get_quantile_bigint")
-kll_sketch_get_quantile_double = _create_wrapper("kll_sketch_get_quantile_double")
-kll_sketch_get_quantile_float = _create_wrapper("kll_sketch_get_quantile_float")
-kll_sketch_get_rank_bigint = _create_wrapper("kll_sketch_get_rank_bigint")
-kll_sketch_get_rank_double = _create_wrapper("kll_sketch_get_rank_double")
-kll_sketch_get_rank_float = _create_wrapper("kll_sketch_get_rank_float")
-kll_sketch_merge_bigint = _create_wrapper("kll_sketch_merge_bigint")
-kll_sketch_merge_double = _create_wrapper("kll_sketch_merge_double")
-kll_sketch_merge_float = _create_wrapper("kll_sketch_merge_float")
-kll_sketch_to_string_bigint = _create_wrapper("kll_sketch_to_string_bigint")
-kll_sketch_to_string_double = _create_wrapper("kll_sketch_to_string_double")
-kll_sketch_to_string_float = _create_wrapper("kll_sketch_to_string_float")
-kurtosis = _create_wrapper("kurtosis")
-lag = _create_wrapper("lag")
-last = _create_wrapper("last")
-last_day = _create_wrapper("last_day")
-last_value = _create_wrapper("last_value")
-lcase = _create_wrapper("lcase")
-lead = _create_wrapper("lead")
-least = _create_wrapper("least")
-left = _create_wrapper("left")
-length = _create_wrapper("length")
-levenshtein = _create_wrapper("levenshtein")
-like = _create_wrapper("like")
-listagg = _create_wrapper("listagg")
-listagg_distinct = _create_wrapper("listagg_distinct")
-ln = _create_wrapper("ln")
-localtimestamp = _create_wrapper("localtimestamp")
-locate = _create_wrapper("locate")
-log = _create_wrapper("log")
-log10 = _create_wrapper("log10")
-log1p = _create_wrapper("log1p")
-log2 = _create_wrapper("log2")
-lower = _create_wrapper("lower")
-lpad = _create_wrapper("lpad")
-ltrim = _create_wrapper("ltrim")
-make_date = _create_wrapper("make_date")
-make_dt_interval = _create_wrapper("make_dt_interval")
-make_interval = _create_wrapper("make_interval")
-make_time = _create_wrapper("make_time")
-make_timestamp = _create_wrapper("make_timestamp")
-make_timestamp_ltz = _create_wrapper("make_timestamp_ltz")
-make_timestamp_ntz = _create_wrapper("make_timestamp_ntz")
-make_valid_utf8 = _create_wrapper("make_valid_utf8")
-make_ym_interval = _create_wrapper("make_ym_interval")
-map_concat = _create_wrapper("map_concat")
-map_contains_key = _create_wrapper("map_contains_key")
-map_entries = _create_wrapper("map_entries")
-map_from_arrays = _create_wrapper("map_from_arrays")
-map_from_entries = _create_wrapper("map_from_entries")
-map_keys = _create_wrapper("map_keys")
-map_values = _create_wrapper("map_values")
-mask = _create_wrapper("mask")
-max_by = _create_wrapper("max_by")
-md5 = _create_wrapper("md5")
-mean = _create_wrapper("mean")
-median = _create_wrapper("median")
-min_by = _create_wrapper("min_by")
-minute = _create_wrapper("minute")
-mode = _create_wrapper("mode")
-monotonically_increasing_id = _create_wrapper("monotonically_increasing_id")
-month = _create_wrapper("month")
-monthname = _create_wrapper("monthname")
-months = _create_wrapper("months")
-months_between = _create_wrapper("months_between")
-named_struct = _create_wrapper("named_struct")
-nanvl = _create_wrapper("nanvl")
-negate = _create_wrapper("negate")
-negative = _create_wrapper("negative")
-next_day = _create_wrapper("next_day")
-now = _create_wrapper("now")
-nth_value = _create_wrapper("nth_value")
-ntile = _create_wrapper("ntile")
-nullif = _create_wrapper("nullif")
-nullifzero = _create_wrapper("nullifzero")
-nvl = _create_wrapper("nvl")
-nvl2 = _create_wrapper("nvl2")
-octet_length = _create_wrapper("octet_length")
-overlay = _create_wrapper("overlay")
-parse_json = _create_wrapper("parse_json")
-parse_url = _create_wrapper("parse_url")
-percent_rank = _create_wrapper("percent_rank")
-percentile = _create_wrapper("percentile")
-percentile_approx = _create_wrapper("percentile_approx")
-pi = _create_wrapper("pi")
-pmod = _create_wrapper("pmod")
-posexplode = _create_wrapper("posexplode")
-posexplode_outer = _create_wrapper("posexplode_outer")
-position = _create_wrapper("position")
-positive = _create_wrapper("positive")
-pow = _create_wrapper("pow")
-power = _create_wrapper("power")
-printf = _create_wrapper("printf")
-product = _create_wrapper("product")
-quarter = _create_wrapper("quarter")
-quote = _create_wrapper("quote")
-radians = _create_wrapper("radians")
-raise_error = _create_wrapper("raise_error")
-rand = _create_wrapper("rand")
-randn = _create_wrapper("randn")
-randstr = _create_wrapper("randstr")
-rank = _create_wrapper("rank")
-reflect = _create_wrapper("reflect")
-regexp = _create_wrapper("regexp")
-regexp_count = _create_wrapper("regexp_count")
-regexp_extract = _create_wrapper("regexp_extract")
-regexp_extract_all = _create_wrapper("regexp_extract_all")
-regexp_instr = _create_wrapper("regexp_instr")
-regexp_like = _create_wrapper("regexp_like")
-regexp_replace = _create_wrapper("regexp_replace")
-regexp_substr = _create_wrapper("regexp_substr")
-regr_avgx = _create_wrapper("regr_avgx")
-regr_avgy = _create_wrapper("regr_avgy")
-regr_count = _create_wrapper("regr_count")
-regr_intercept = _create_wrapper("regr_intercept")
-regr_r2 = _create_wrapper("regr_r2")
-regr_slope = _create_wrapper("regr_slope")
-regr_sxx = _create_wrapper("regr_sxx")
-regr_sxy = _create_wrapper("regr_sxy")
-regr_syy = _create_wrapper("regr_syy")
-repeat = _create_wrapper("repeat")
-replace = _create_wrapper("replace")
-reverse = _create_wrapper("reverse")
-right = _create_wrapper("right")
-rint = _create_wrapper("rint")
-rlike = _create_wrapper("rlike")
-round = _create_wrapper("round")
-row_number = _create_wrapper("row_number")
-rpad = _create_wrapper("rpad")
-rtrim = _create_wrapper("rtrim")
-schema_of_csv = _create_wrapper("schema_of_csv")
-schema_of_json = _create_wrapper("schema_of_json")
-schema_of_variant = _create_wrapper("schema_of_variant")
-schema_of_variant_agg = _create_wrapper("schema_of_variant_agg")
-schema_of_xml = _create_wrapper("schema_of_xml")
-sec = _create_wrapper("sec")
-second = _create_wrapper("second")
-sentences = _create_wrapper("sentences")
-sequence = _create_wrapper("sequence")
-session_user = _create_wrapper("session_user")
-session_window = _create_wrapper("session_window")
-sha = _create_wrapper("sha")
-sha1 = _create_wrapper("sha1")
-shiftLeft = _create_wrapper("shiftLeft")
-shiftRight = _create_wrapper("shiftRight")
-shiftRightUnsigned = _create_wrapper("shiftRightUnsigned")
-shiftleft = _create_wrapper("shiftleft")
-shiftright = _create_wrapper("shiftright")
-shiftrightunsigned = _create_wrapper("shiftrightunsigned")
-shuffle = _create_wrapper("shuffle")
-sign = _create_wrapper("sign")
-signum = _create_wrapper("signum")
-sin = _create_wrapper("sin")
-sinh = _create_wrapper("sinh")
-size = _create_wrapper("size")
-skewness = _create_wrapper("skewness")
-slice = _create_wrapper("slice")
-some = _create_wrapper("some")
-sort_array = _create_wrapper("sort_array")
-soundex = _create_wrapper("soundex")
-spark_partition_id = _create_wrapper("spark_partition_id")
-split = _create_wrapper("split")
-split_part = _create_wrapper("split_part")
-sqrt = _create_wrapper("sqrt")
-st_asbinary = _create_wrapper("st_asbinary")
-st_geogfromwkb = _create_wrapper("st_geogfromwkb")
-st_geomfromwkb = _create_wrapper("st_geomfromwkb")
-st_setsrid = _create_wrapper("st_setsrid")
-st_srid = _create_wrapper("st_srid")
-stack = _create_wrapper("stack")
-startswith = _create_wrapper("startswith")
-std = _create_wrapper("std")
-stddev = _create_wrapper("stddev")
-stddev_pop = _create_wrapper("stddev_pop")
-stddev_samp = _create_wrapper("stddev_samp")
-str_to_map = _create_wrapper("str_to_map")
-string_agg = _create_wrapper("string_agg")
-string_agg_distinct = _create_wrapper("string_agg_distinct")
-struct = _create_wrapper("struct")
-substr = _create_wrapper("substr")
-substring = _create_wrapper("substring")
-substring_index = _create_wrapper("substring_index")
-sumDistinct = _create_wrapper("sumDistinct")
-sum_distinct = _create_wrapper("sum_distinct")
-tan = _create_wrapper("tan")
-tanh = _create_wrapper("tanh")
-theta_difference = _create_wrapper("theta_difference")
-theta_intersection = _create_wrapper("theta_intersection")
-theta_intersection_agg = _create_wrapper("theta_intersection_agg")
-theta_sketch_agg = _create_wrapper("theta_sketch_agg")
-theta_sketch_estimate = _create_wrapper("theta_sketch_estimate")
-theta_union = _create_wrapper("theta_union")
-theta_union_agg = _create_wrapper("theta_union_agg")
-time_bucket = _create_wrapper("time_bucket")
-time_diff = _create_wrapper("time_diff")
-time_from_micros = _create_wrapper("time_from_micros")
-time_from_millis = _create_wrapper("time_from_millis")
-time_from_seconds = _create_wrapper("time_from_seconds")
-time_to_micros = _create_wrapper("time_to_micros")
-time_to_millis = _create_wrapper("time_to_millis")
-time_to_seconds = _create_wrapper("time_to_seconds")
-time_trunc = _create_wrapper("time_trunc")
-timestamp_add = _create_wrapper("timestamp_add")
-timestamp_diff = _create_wrapper("timestamp_diff")
-timestamp_micros = _create_wrapper("timestamp_micros")
-timestamp_millis = _create_wrapper("timestamp_millis")
+def java_method(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("java_method", _a)
+def json_array_length(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("json_array_length", _a)
+def json_object_keys(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("json_object_keys", _a)
+def json_tuple(col, *fields):
+ _a = []
+ _a.append(col)
+ _a.extend(fields)
+ return _dispatch("json_tuple", _a)
+def kll_merge_agg_bigint(col, k=_UNSET):
+ _a = []
+ _a.append(col)
+ if k is not _UNSET: _a.append(k)
+ return _dispatch("kll_merge_agg_bigint", _a)
+def kll_merge_agg_double(col, k=_UNSET):
+ _a = []
+ _a.append(col)
+ if k is not _UNSET: _a.append(k)
+ return _dispatch("kll_merge_agg_double", _a)
+def kll_merge_agg_float(col, k=_UNSET):
+ _a = []
+ _a.append(col)
+ if k is not _UNSET: _a.append(k)
+ return _dispatch("kll_merge_agg_float", _a)
+def kll_sketch_agg_bigint(col, k=_UNSET):
+ _a = []
+ _a.append(col)
+ if k is not _UNSET: _a.append(k)
+ return _dispatch("kll_sketch_agg_bigint", _a)
+def kll_sketch_agg_double(col, k=_UNSET):
+ _a = []
+ _a.append(col)
+ if k is not _UNSET: _a.append(k)
+ return _dispatch("kll_sketch_agg_double", _a)
+def kll_sketch_agg_float(col, k=_UNSET):
+ _a = []
+ _a.append(col)
+ if k is not _UNSET: _a.append(k)
+ return _dispatch("kll_sketch_agg_float", _a)
+def kll_sketch_get_n_bigint(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("kll_sketch_get_n_bigint", _a)
+def kll_sketch_get_n_double(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("kll_sketch_get_n_double", _a)
+def kll_sketch_get_n_float(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("kll_sketch_get_n_float", _a)
+def kll_sketch_get_quantile_bigint(sketch, rank):
+ _a = []
+ _a.append(sketch)
+ _a.append(rank)
+ return _dispatch("kll_sketch_get_quantile_bigint", _a)
+def kll_sketch_get_quantile_double(sketch, rank):
+ _a = []
+ _a.append(sketch)
+ _a.append(rank)
+ return _dispatch("kll_sketch_get_quantile_double", _a)
+def kll_sketch_get_quantile_float(sketch, rank):
+ _a = []
+ _a.append(sketch)
+ _a.append(rank)
+ return _dispatch("kll_sketch_get_quantile_float", _a)
+def kll_sketch_get_rank_bigint(sketch, quantile):
+ _a = []
+ _a.append(sketch)
+ _a.append(quantile)
+ return _dispatch("kll_sketch_get_rank_bigint", _a)
+def kll_sketch_get_rank_double(sketch, quantile):
+ _a = []
+ _a.append(sketch)
+ _a.append(quantile)
+ return _dispatch("kll_sketch_get_rank_double", _a)
+def kll_sketch_get_rank_float(sketch, quantile):
+ _a = []
+ _a.append(sketch)
+ _a.append(quantile)
+ return _dispatch("kll_sketch_get_rank_float", _a)
+def kll_sketch_merge_bigint(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("kll_sketch_merge_bigint", _a)
+def kll_sketch_merge_double(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("kll_sketch_merge_double", _a)
+def kll_sketch_merge_float(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("kll_sketch_merge_float", _a)
+def kll_sketch_to_string_bigint(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("kll_sketch_to_string_bigint", _a)
+def kll_sketch_to_string_double(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("kll_sketch_to_string_double", _a)
+def kll_sketch_to_string_float(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("kll_sketch_to_string_float", _a)
+def kurtosis(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("kurtosis", _a)
+def lag(col, offset=_UNSET, default=_UNSET):
+ _a = []
+ _a.append(col)
+ if offset is not _UNSET: _a.append(offset)
+ if default is not _UNSET: _a.append(default)
+ return _dispatch("lag", _a)
+def last(col, ignorenulls=_UNSET):
+ _a = []
+ _a.append(col)
+ if ignorenulls is not _UNSET: _a.append(ignorenulls)
+ return _dispatch("last", _a)
+def last_day(date):
+ _a = []
+ _a.append(date)
+ return _dispatch("last_day", _a)
+def last_value(col, ignoreNulls=_UNSET):
+ _a = []
+ _a.append(col)
+ if ignoreNulls is not _UNSET: _a.append(ignoreNulls)
+ return _dispatch("last_value", _a)
+def lcase(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("lcase", _a)
+def lead(col, offset=_UNSET, default=_UNSET):
+ _a = []
+ _a.append(col)
+ if offset is not _UNSET: _a.append(offset)
+ if default is not _UNSET: _a.append(default)
+ return _dispatch("lead", _a)
+def least(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("least", _a)
+def left(str, len):
+ _a = []
+ _a.append(str)
+ _a.append(len)
+ return _dispatch("left", _a)
+def length(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("length", _a)
+def levenshtein(left, right, threshold=_UNSET):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ if threshold is not _UNSET: _a.append(threshold)
+ return _dispatch("levenshtein", _a)
+def like(str, pattern, escapeChar=_UNSET):
+ _a = []
+ _a.append(str)
+ _a.append(pattern)
+ if escapeChar is not _UNSET: _a.append(escapeChar)
+ return _dispatch("like", _a)
+def listagg(col, delimiter=_UNSET):
+ _a = []
+ _a.append(col)
+ if delimiter is not _UNSET: _a.append(delimiter)
+ return _dispatch("listagg", _a)
+def listagg_distinct(col, delimiter=_UNSET):
+ _a = []
+ _a.append(col)
+ if delimiter is not _UNSET: _a.append(delimiter)
+ return _dispatch("listagg_distinct", _a)
+def ln(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("ln", _a)
+def localtimestamp():
+ _a = []
+ return _dispatch("localtimestamp", _a)
+def locate(substr, str, pos=_UNSET):
+ _a = []
+ _a.append(substr)
+ _a.append(str)
+ if pos is not _UNSET: _a.append(pos)
+ return _dispatch("locate", _a)
+def log(arg1, arg2=_UNSET):
+ _a = []
+ _a.append(arg1)
+ if arg2 is not _UNSET: _a.append(arg2)
+ return _dispatch("log", _a)
+def log10(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("log10", _a)
+def log1p(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("log1p", _a)
+def log2(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("log2", _a)
+def lower(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("lower", _a)
+def lpad(col, len, pad):
+ _a = []
+ _a.append(col)
+ _a.append(len)
+ _a.append(pad)
+ return _dispatch("lpad", _a)
+def ltrim(col, trim=_UNSET):
+ _a = []
+ _a.append(col)
+ if trim is not _UNSET: _a.append(trim)
+ return _dispatch("ltrim", _a)
+def make_date(year, month, day):
+ _a = []
+ _a.append(year)
+ _a.append(month)
+ _a.append(day)
+ return _dispatch("make_date", _a)
+def make_dt_interval(days=_UNSET, hours=_UNSET, mins=_UNSET, secs=_UNSET):
+ _a = []
+ if days is not _UNSET: _a.append(days)
+ if hours is not _UNSET: _a.append(hours)
+ if mins is not _UNSET: _a.append(mins)
+ if secs is not _UNSET: _a.append(secs)
+ return _dispatch("make_dt_interval", _a)
+def make_interval(years=_UNSET, months=_UNSET, weeks=_UNSET, days=_UNSET, hours=_UNSET, mins=_UNSET, secs=_UNSET):
+ _a = []
+ if years is not _UNSET: _a.append(years)
+ if months is not _UNSET: _a.append(months)
+ if weeks is not _UNSET: _a.append(weeks)
+ if days is not _UNSET: _a.append(days)
+ if hours is not _UNSET: _a.append(hours)
+ if mins is not _UNSET: _a.append(mins)
+ if secs is not _UNSET: _a.append(secs)
+ return _dispatch("make_interval", _a)
+def make_time(hour, minute, second):
+ _a = []
+ _a.append(hour)
+ _a.append(minute)
+ _a.append(second)
+ return _dispatch("make_time", _a)
+def make_timestamp(years=_UNSET, months=_UNSET, days=_UNSET, hours=_UNSET, mins=_UNSET, secs=_UNSET, timezone=_UNSET, date=_UNSET, time=_UNSET):
+ _a = []
+ if years is not _UNSET: _a.append(years)
+ if months is not _UNSET: _a.append(months)
+ if days is not _UNSET: _a.append(days)
+ if hours is not _UNSET: _a.append(hours)
+ if mins is not _UNSET: _a.append(mins)
+ if secs is not _UNSET: _a.append(secs)
+ if timezone is not _UNSET: _a.append(timezone)
+ if date is not _UNSET: _a.append(date)
+ if time is not _UNSET: _a.append(time)
+ return _dispatch("make_timestamp", _a)
+def make_timestamp_ltz(years, months, days, hours, mins, secs, timezone=_UNSET):
+ _a = []
+ _a.append(years)
+ _a.append(months)
+ _a.append(days)
+ _a.append(hours)
+ _a.append(mins)
+ _a.append(secs)
+ if timezone is not _UNSET: _a.append(timezone)
+ return _dispatch("make_timestamp_ltz", _a)
+def make_timestamp_ntz(years=_UNSET, months=_UNSET, days=_UNSET, hours=_UNSET, mins=_UNSET, secs=_UNSET, date=_UNSET, time=_UNSET):
+ _a = []
+ if years is not _UNSET: _a.append(years)
+ if months is not _UNSET: _a.append(months)
+ if days is not _UNSET: _a.append(days)
+ if hours is not _UNSET: _a.append(hours)
+ if mins is not _UNSET: _a.append(mins)
+ if secs is not _UNSET: _a.append(secs)
+ if date is not _UNSET: _a.append(date)
+ if time is not _UNSET: _a.append(time)
+ return _dispatch("make_timestamp_ntz", _a)
+def make_valid_utf8(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("make_valid_utf8", _a)
+def make_ym_interval(years=_UNSET, months=_UNSET):
+ _a = []
+ if years is not _UNSET: _a.append(years)
+ if months is not _UNSET: _a.append(months)
+ return _dispatch("make_ym_interval", _a)
+def map_concat(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("map_concat", _a)
+def map_contains_key(col, value):
+ _a = []
+ _a.append(col)
+ _a.append(value)
+ return _dispatch("map_contains_key", _a)
+def map_entries(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("map_entries", _a)
+def map_from_arrays(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("map_from_arrays", _a)
+def map_from_entries(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("map_from_entries", _a)
+def map_keys(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("map_keys", _a)
+def map_values(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("map_values", _a)
+def mask(col, upperChar=_UNSET, lowerChar=_UNSET, digitChar=_UNSET, otherChar=_UNSET):
+ _a = []
+ _a.append(col)
+ if upperChar is not _UNSET: _a.append(upperChar)
+ if lowerChar is not _UNSET: _a.append(lowerChar)
+ if digitChar is not _UNSET: _a.append(digitChar)
+ if otherChar is not _UNSET: _a.append(otherChar)
+ return _dispatch("mask", _a)
+def max_by(col, ord, k=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(ord)
+ if k is not _UNSET: _a.append(k)
+ return _dispatch("max_by", _a)
+def md5(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("md5", _a)
+def mean(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("mean", _a)
+def median(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("median", _a)
+def min_by(col, ord, k=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(ord)
+ if k is not _UNSET: _a.append(k)
+ return _dispatch("min_by", _a)
+def minute(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("minute", _a)
+def mode(col, deterministic=_UNSET):
+ _a = []
+ _a.append(col)
+ if deterministic is not _UNSET: _a.append(deterministic)
+ return _dispatch("mode", _a)
+def monotonically_increasing_id():
+ _a = []
+ return _dispatch("monotonically_increasing_id", _a)
+def month(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("month", _a)
+def monthname(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("monthname", _a)
+def months(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("months", _a)
+def months_between(date1, date2, roundOff=_UNSET):
+ _a = []
+ _a.append(date1)
+ _a.append(date2)
+ if roundOff is not _UNSET: _a.append(roundOff)
+ return _dispatch("months_between", _a)
+def named_struct(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("named_struct", _a)
+def nanvl(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("nanvl", _a)
+def negate(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("negate", _a)
+def negative(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("negative", _a)
+def next_day(date, dayOfWeek):
+ _a = []
+ _a.append(date)
+ _a.append(dayOfWeek)
+ return _dispatch("next_day", _a)
+def now():
+ _a = []
+ return _dispatch("now", _a)
+def nth_value(col, offset, ignoreNulls=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(offset)
+ if ignoreNulls is not _UNSET: _a.append(ignoreNulls)
+ return _dispatch("nth_value", _a)
+def ntile(n):
+ _a = []
+ _a.append(n)
+ return _dispatch("ntile", _a)
+def nullif(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("nullif", _a)
+def nullifzero(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("nullifzero", _a)
+def nvl(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("nvl", _a)
+def nvl2(col1, col2, col3):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ _a.append(col3)
+ return _dispatch("nvl2", _a)
+def octet_length(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("octet_length", _a)
+def overlay(src, replace, pos, len=_UNSET):
+ _a = []
+ _a.append(src)
+ _a.append(replace)
+ _a.append(pos)
+ if len is not _UNSET: _a.append(len)
+ return _dispatch("overlay", _a)
+def parse_json(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("parse_json", _a)
+def parse_url(url, partToExtract, key=_UNSET):
+ _a = []
+ _a.append(url)
+ _a.append(partToExtract)
+ if key is not _UNSET: _a.append(key)
+ return _dispatch("parse_url", _a)
+def percent_rank():
+ _a = []
+ return _dispatch("percent_rank", _a)
+def percentile(col, percentage, frequency=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(percentage)
+ if frequency is not _UNSET: _a.append(frequency)
+ return _dispatch("percentile", _a)
+def percentile_approx(col, percentage, accuracy=_UNSET):
+ _a = []
+ _a.append(col)
+ _a.append(percentage)
+ if accuracy is not _UNSET: _a.append(accuracy)
+ return _dispatch("percentile_approx", _a)
+def pi():
+ _a = []
+ return _dispatch("pi", _a)
+def pmod(dividend, divisor):
+ _a = []
+ _a.append(dividend)
+ _a.append(divisor)
+ return _dispatch("pmod", _a)
+def posexplode(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("posexplode", _a)
+def posexplode_outer(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("posexplode_outer", _a)
+def position(substr, str, start=_UNSET):
+ _a = []
+ _a.append(substr)
+ _a.append(str)
+ if start is not _UNSET: _a.append(start)
+ return _dispatch("position", _a)
+def positive(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("positive", _a)
+def pow(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("pow", _a)
+def power(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("power", _a)
+def printf(format, *cols):
+ _a = []
+ _a.append(format)
+ _a.extend(cols)
+ return _dispatch("printf", _a)
+def product(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("product", _a)
+def quarter(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("quarter", _a)
+def quote(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("quote", _a)
+def radians(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("radians", _a)
+def raise_error(errMsg):
+ _a = []
+ _a.append(errMsg)
+ return _dispatch("raise_error", _a)
+def rand(seed=_UNSET):
+ _a = []
+ if seed is not _UNSET: _a.append(seed)
+ return _dispatch("rand", _a)
+def randn(seed=_UNSET):
+ _a = []
+ if seed is not _UNSET: _a.append(seed)
+ return _dispatch("randn", _a)
+def randstr(length, seed=_UNSET):
+ _a = []
+ _a.append(length)
+ if seed is not _UNSET: _a.append(seed)
+ return _dispatch("randstr", _a)
+def rank():
+ _a = []
+ return _dispatch("rank", _a)
+def reflect(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("reflect", _a)
+def regexp(str, regexp):
+ _a = []
+ _a.append(str)
+ _a.append(regexp)
+ return _dispatch("regexp", _a)
+def regexp_count(str, regexp):
+ _a = []
+ _a.append(str)
+ _a.append(regexp)
+ return _dispatch("regexp_count", _a)
+def regexp_extract(str, pattern, idx):
+ _a = []
+ _a.append(str)
+ _a.append(pattern)
+ _a.append(idx)
+ return _dispatch("regexp_extract", _a)
+def regexp_extract_all(str, regexp, idx=_UNSET):
+ _a = []
+ _a.append(str)
+ _a.append(regexp)
+ if idx is not _UNSET: _a.append(idx)
+ return _dispatch("regexp_extract_all", _a)
+def regexp_instr(str, regexp, idx=_UNSET):
+ _a = []
+ _a.append(str)
+ _a.append(regexp)
+ if idx is not _UNSET: _a.append(idx)
+ return _dispatch("regexp_instr", _a)
+def regexp_like(str, regexp):
+ _a = []
+ _a.append(str)
+ _a.append(regexp)
+ return _dispatch("regexp_like", _a)
+def regexp_replace(string, pattern, replacement):
+ _a = []
+ _a.append(string)
+ _a.append(pattern)
+ _a.append(replacement)
+ return _dispatch("regexp_replace", _a)
+def regexp_substr(str, regexp):
+ _a = []
+ _a.append(str)
+ _a.append(regexp)
+ return _dispatch("regexp_substr", _a)
+def regr_avgx(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_avgx", _a)
+def regr_avgy(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_avgy", _a)
+def regr_count(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_count", _a)
+def regr_intercept(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_intercept", _a)
+def regr_r2(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_r2", _a)
+def regr_slope(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_slope", _a)
+def regr_sxx(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_sxx", _a)
+def regr_sxy(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_sxy", _a)
+def regr_syy(y, x):
+ _a = []
+ _a.append(y)
+ _a.append(x)
+ return _dispatch("regr_syy", _a)
+def repeat(col, n):
+ _a = []
+ _a.append(col)
+ _a.append(n)
+ return _dispatch("repeat", _a)
+def replace(src, search, replace=_UNSET):
+ _a = []
+ _a.append(src)
+ _a.append(search)
+ if replace is not _UNSET: _a.append(replace)
+ return _dispatch("replace", _a)
+def reverse(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("reverse", _a)
+def right(str, len):
+ _a = []
+ _a.append(str)
+ _a.append(len)
+ return _dispatch("right", _a)
+def rint(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("rint", _a)
+def rlike(str, regexp):
+ _a = []
+ _a.append(str)
+ _a.append(regexp)
+ return _dispatch("rlike", _a)
+def round(col, scale=_UNSET):
+ _a = []
+ _a.append(col)
+ if scale is not _UNSET: _a.append(scale)
+ return _dispatch("round", _a)
+def row_number():
+ _a = []
+ return _dispatch("row_number", _a)
+def rpad(col, len, pad):
+ _a = []
+ _a.append(col)
+ _a.append(len)
+ _a.append(pad)
+ return _dispatch("rpad", _a)
+def rtrim(col, trim=_UNSET):
+ _a = []
+ _a.append(col)
+ if trim is not _UNSET: _a.append(trim)
+ return _dispatch("rtrim", _a)
+def schema_of_csv(csv, options=_UNSET):
+ _a = []
+ _a.append(csv)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("schema_of_csv", _a)
+def schema_of_json(json, options=_UNSET):
+ _a = []
+ _a.append(json)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("schema_of_json", _a)
+def schema_of_variant(v):
+ _a = []
+ _a.append(v)
+ return _dispatch("schema_of_variant", _a)
+def schema_of_variant_agg(v):
+ _a = []
+ _a.append(v)
+ return _dispatch("schema_of_variant_agg", _a)
+def schema_of_xml(xml, options=_UNSET):
+ _a = []
+ _a.append(xml)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("schema_of_xml", _a)
+def sec(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sec", _a)
+def second(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("second", _a)
+def sentences(string, language=_UNSET, country=_UNSET):
+ _a = []
+ _a.append(string)
+ if language is not _UNSET: _a.append(language)
+ if country is not _UNSET: _a.append(country)
+ return _dispatch("sentences", _a)
+def sequence(start, stop, step=_UNSET):
+ _a = []
+ _a.append(start)
+ _a.append(stop)
+ if step is not _UNSET: _a.append(step)
+ return _dispatch("sequence", _a)
+def session_user():
+ _a = []
+ return _dispatch("session_user", _a)
+def session_window(timeColumn, gapDuration):
+ _a = []
+ _a.append(timeColumn)
+ _a.append(gapDuration)
+ return _dispatch("session_window", _a)
+def sha(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sha", _a)
+def sha1(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sha1", _a)
+def shiftLeft(col, numBits):
+ _a = []
+ _a.append(col)
+ _a.append(numBits)
+ return _dispatch("shiftLeft", _a)
+def shiftRight(col, numBits):
+ _a = []
+ _a.append(col)
+ _a.append(numBits)
+ return _dispatch("shiftRight", _a)
+def shiftRightUnsigned(col, numBits):
+ _a = []
+ _a.append(col)
+ _a.append(numBits)
+ return _dispatch("shiftRightUnsigned", _a)
+def shiftleft(col, numBits):
+ _a = []
+ _a.append(col)
+ _a.append(numBits)
+ return _dispatch("shiftleft", _a)
+def shiftright(col, numBits):
+ _a = []
+ _a.append(col)
+ _a.append(numBits)
+ return _dispatch("shiftright", _a)
+def shiftrightunsigned(col, numBits):
+ _a = []
+ _a.append(col)
+ _a.append(numBits)
+ return _dispatch("shiftrightunsigned", _a)
+def shuffle(col, seed=_UNSET):
+ _a = []
+ _a.append(col)
+ if seed is not _UNSET: _a.append(seed)
+ return _dispatch("shuffle", _a)
+def sign(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sign", _a)
+def signum(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("signum", _a)
+def sin(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sin", _a)
+def sinh(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sinh", _a)
+def size(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("size", _a)
+def skewness(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("skewness", _a)
+def slice(x, start, length):
+ _a = []
+ _a.append(x)
+ _a.append(start)
+ _a.append(length)
+ return _dispatch("slice", _a)
+def some(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("some", _a)
+def sort_array(col, asc=_UNSET):
+ _a = []
+ _a.append(col)
+ if asc is not _UNSET: _a.append(asc)
+ return _dispatch("sort_array", _a)
+def soundex(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("soundex", _a)
+def spark_partition_id():
+ _a = []
+ return _dispatch("spark_partition_id", _a)
+def split(str, pattern, limit=_UNSET):
+ _a = []
+ _a.append(str)
+ _a.append(pattern)
+ if limit is not _UNSET: _a.append(limit)
+ return _dispatch("split", _a)
+def split_part(src, delimiter, partNum):
+ _a = []
+ _a.append(src)
+ _a.append(delimiter)
+ _a.append(partNum)
+ return _dispatch("split_part", _a)
+def sqrt(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sqrt", _a)
+def st_asbinary(geo, endianness=_UNSET):
+ _a = []
+ _a.append(geo)
+ if endianness is not _UNSET: _a.append(endianness)
+ return _dispatch("st_asbinary", _a)
+def st_geogfromwkb(wkb):
+ _a = []
+ _a.append(wkb)
+ return _dispatch("st_geogfromwkb", _a)
+def st_geomfromwkb(wkb, srid=_UNSET):
+ _a = []
+ _a.append(wkb)
+ if srid is not _UNSET: _a.append(srid)
+ return _dispatch("st_geomfromwkb", _a)
+def st_setsrid(geo, srid):
+ _a = []
+ _a.append(geo)
+ _a.append(srid)
+ return _dispatch("st_setsrid", _a)
+def st_srid(geo):
+ _a = []
+ _a.append(geo)
+ return _dispatch("st_srid", _a)
+def stack(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("stack", _a)
+def startswith(str, prefix):
+ _a = []
+ _a.append(str)
+ _a.append(prefix)
+ return _dispatch("startswith", _a)
+def std(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("std", _a)
+def stddev(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("stddev", _a)
+def stddev_pop(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("stddev_pop", _a)
+def stddev_samp(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("stddev_samp", _a)
+def str_to_map(text, pairDelim=_UNSET, keyValueDelim=_UNSET):
+ _a = []
+ _a.append(text)
+ if pairDelim is not _UNSET: _a.append(pairDelim)
+ if keyValueDelim is not _UNSET: _a.append(keyValueDelim)
+ return _dispatch("str_to_map", _a)
+def string_agg(col, delimiter=_UNSET):
+ _a = []
+ _a.append(col)
+ if delimiter is not _UNSET: _a.append(delimiter)
+ return _dispatch("string_agg", _a)
+def string_agg_distinct(col, delimiter=_UNSET):
+ _a = []
+ _a.append(col)
+ if delimiter is not _UNSET: _a.append(delimiter)
+ return _dispatch("string_agg_distinct", _a)
+def struct(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("struct", _a)
+def substr(str, pos, len=_UNSET):
+ _a = []
+ _a.append(str)
+ _a.append(pos)
+ if len is not _UNSET: _a.append(len)
+ return _dispatch("substr", _a)
+def substring(str, pos, len):
+ _a = []
+ _a.append(str)
+ _a.append(pos)
+ _a.append(len)
+ return _dispatch("substring", _a)
+def substring_index(str, delim, count):
+ _a = []
+ _a.append(str)
+ _a.append(delim)
+ _a.append(count)
+ return _dispatch("substring_index", _a)
+def sumDistinct(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sumDistinct", _a)
+def sum_distinct(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("sum_distinct", _a)
+def tan(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("tan", _a)
+def tanh(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("tanh", _a)
+def theta_difference(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("theta_difference", _a)
+def theta_intersection(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("theta_intersection", _a)
+def theta_intersection_agg(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("theta_intersection_agg", _a)
+def theta_sketch_agg(col, lgNomEntries=_UNSET):
+ _a = []
+ _a.append(col)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ return _dispatch("theta_sketch_agg", _a)
+def theta_sketch_estimate(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("theta_sketch_estimate", _a)
+def theta_union(col1, col2, lgNomEntries=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ return _dispatch("theta_union", _a)
+def theta_union_agg(col, lgNomEntries=_UNSET):
+ _a = []
+ _a.append(col)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ return _dispatch("theta_union_agg", _a)
+def time_bucket(bucket_size, ts, origin=_UNSET):
+ _a = []
+ _a.append(bucket_size)
+ _a.append(ts)
+ if origin is not _UNSET: _a.append(origin)
+ return _dispatch("time_bucket", _a)
+def time_diff(unit, start, end):
+ _a = []
+ _a.append(unit)
+ _a.append(start)
+ _a.append(end)
+ return _dispatch("time_diff", _a)
+def time_from_micros(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("time_from_micros", _a)
+def time_from_millis(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("time_from_millis", _a)
+def time_from_seconds(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("time_from_seconds", _a)
+def time_to_micros(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("time_to_micros", _a)
+def time_to_millis(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("time_to_millis", _a)
+def time_to_seconds(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("time_to_seconds", _a)
+def time_trunc(unit, time):
+ _a = []
+ _a.append(unit)
+ _a.append(time)
+ return _dispatch("time_trunc", _a)
+def timestamp_add(unit, quantity, ts):
+ _a = []
+ _a.append(unit)
+ _a.append(quantity)
+ _a.append(ts)
+ return _dispatch("timestamp_add", _a)
+def timestamp_diff(unit, start, end):
+ _a = []
+ _a.append(unit)
+ _a.append(start)
+ _a.append(end)
+ return _dispatch("timestamp_diff", _a)
+def timestamp_micros(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("timestamp_micros", _a)
+def timestamp_millis(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("timestamp_millis", _a)
timestamp_nanos = _create_wrapper("timestamp_nanos")
-timestamp_seconds = _create_wrapper("timestamp_seconds")
-toDegrees = _create_wrapper("toDegrees")
-toRadians = _create_wrapper("toRadians")
+def timestamp_seconds(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("timestamp_seconds", _a)
+def toDegrees(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("toDegrees", _a)
+def toRadians(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("toRadians", _a)
to_avro = _create_wrapper("to_avro")
-to_binary = _create_wrapper("to_binary")
-to_char = _create_wrapper("to_char")
-to_csv = _create_wrapper("to_csv")
-to_date = _create_wrapper("to_date")
-to_json = _create_wrapper("to_json")
-to_number = _create_wrapper("to_number")
-to_time = _create_wrapper("to_time")
-to_timestamp = _create_wrapper("to_timestamp")
-to_timestamp_ltz = _create_wrapper("to_timestamp_ltz")
-to_timestamp_ntz = _create_wrapper("to_timestamp_ntz")
-to_unix_timestamp = _create_wrapper("to_unix_timestamp")
-to_utc_timestamp = _create_wrapper("to_utc_timestamp")
-to_varchar = _create_wrapper("to_varchar")
-to_variant_object = _create_wrapper("to_variant_object")
-to_xml = _create_wrapper("to_xml")
-translate = _create_wrapper("translate")
-trim = _create_wrapper("trim")
-trunc = _create_wrapper("trunc")
-try_add = _create_wrapper("try_add")
-try_aes_decrypt = _create_wrapper("try_aes_decrypt")
-try_avg = _create_wrapper("try_avg")
-try_divide = _create_wrapper("try_divide")
-try_element_at = _create_wrapper("try_element_at")
-try_make_interval = _create_wrapper("try_make_interval")
-try_make_timestamp = _create_wrapper("try_make_timestamp")
-try_make_timestamp_ltz = _create_wrapper("try_make_timestamp_ltz")
-try_make_timestamp_ntz = _create_wrapper("try_make_timestamp_ntz")
-try_mod = _create_wrapper("try_mod")
-try_multiply = _create_wrapper("try_multiply")
-try_parse_json = _create_wrapper("try_parse_json")
-try_parse_url = _create_wrapper("try_parse_url")
-try_reflect = _create_wrapper("try_reflect")
-try_subtract = _create_wrapper("try_subtract")
-try_sum = _create_wrapper("try_sum")
-try_to_binary = _create_wrapper("try_to_binary")
-try_to_date = _create_wrapper("try_to_date")
-try_to_number = _create_wrapper("try_to_number")
-try_to_time = _create_wrapper("try_to_time")
-try_to_timestamp = _create_wrapper("try_to_timestamp")
-try_url_decode = _create_wrapper("try_url_decode")
-try_validate_utf8 = _create_wrapper("try_validate_utf8")
+def to_binary(col, format=_UNSET):
+ _a = []
+ _a.append(col)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("to_binary", _a)
+def to_char(col, format):
+ _a = []
+ _a.append(col)
+ _a.append(format)
+ return _dispatch("to_char", _a)
+def to_csv(col, options=_UNSET):
+ _a = []
+ _a.append(col)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("to_csv", _a)
+def to_date(col, format=_UNSET):
+ _a = []
+ _a.append(col)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("to_date", _a)
+def to_json(col, options=_UNSET):
+ _a = []
+ _a.append(col)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("to_json", _a)
+def to_number(col, format):
+ _a = []
+ _a.append(col)
+ _a.append(format)
+ return _dispatch("to_number", _a)
+def to_time(str, format=_UNSET):
+ _a = []
+ _a.append(str)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("to_time", _a)
+def to_timestamp(col, format=_UNSET):
+ _a = []
+ _a.append(col)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("to_timestamp", _a)
+def to_timestamp_ltz(timestamp, format=_UNSET):
+ _a = []
+ _a.append(timestamp)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("to_timestamp_ltz", _a)
+def to_timestamp_ntz(timestamp, format=_UNSET):
+ _a = []
+ _a.append(timestamp)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("to_timestamp_ntz", _a)
+def to_unix_timestamp(timestamp, format=_UNSET):
+ _a = []
+ _a.append(timestamp)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("to_unix_timestamp", _a)
+def to_utc_timestamp(timestamp, tz):
+ _a = []
+ _a.append(timestamp)
+ _a.append(tz)
+ return _dispatch("to_utc_timestamp", _a)
+def to_varchar(col, format):
+ _a = []
+ _a.append(col)
+ _a.append(format)
+ return _dispatch("to_varchar", _a)
+def to_variant_object(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("to_variant_object", _a)
+def to_xml(col, options=_UNSET):
+ _a = []
+ _a.append(col)
+ if options is not _UNSET: _a.append(options)
+ return _dispatch("to_xml", _a)
+def translate(srcCol, matching, replace):
+ _a = []
+ _a.append(srcCol)
+ _a.append(matching)
+ _a.append(replace)
+ return _dispatch("translate", _a)
+def trim(col, trim=_UNSET):
+ _a = []
+ _a.append(col)
+ if trim is not _UNSET: _a.append(trim)
+ return _dispatch("trim", _a)
+def trunc(date, format):
+ _a = []
+ _a.append(date)
+ _a.append(format)
+ return _dispatch("trunc", _a)
+def try_add(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("try_add", _a)
+def try_aes_decrypt(input, key, mode=_UNSET, padding=_UNSET, aad=_UNSET):
+ _a = []
+ _a.append(input)
+ _a.append(key)
+ if mode is not _UNSET: _a.append(mode)
+ if padding is not _UNSET: _a.append(padding)
+ if aad is not _UNSET: _a.append(aad)
+ return _dispatch("try_aes_decrypt", _a)
+def try_avg(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("try_avg", _a)
+def try_divide(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("try_divide", _a)
+def try_element_at(col, extraction):
+ _a = []
+ _a.append(col)
+ _a.append(extraction)
+ return _dispatch("try_element_at", _a)
+def try_make_interval(years=_UNSET, months=_UNSET, weeks=_UNSET, days=_UNSET, hours=_UNSET, mins=_UNSET, secs=_UNSET):
+ _a = []
+ if years is not _UNSET: _a.append(years)
+ if months is not _UNSET: _a.append(months)
+ if weeks is not _UNSET: _a.append(weeks)
+ if days is not _UNSET: _a.append(days)
+ if hours is not _UNSET: _a.append(hours)
+ if mins is not _UNSET: _a.append(mins)
+ if secs is not _UNSET: _a.append(secs)
+ return _dispatch("try_make_interval", _a)
+def try_make_timestamp(years=_UNSET, months=_UNSET, days=_UNSET, hours=_UNSET, mins=_UNSET, secs=_UNSET, timezone=_UNSET, date=_UNSET, time=_UNSET):
+ _a = []
+ if years is not _UNSET: _a.append(years)
+ if months is not _UNSET: _a.append(months)
+ if days is not _UNSET: _a.append(days)
+ if hours is not _UNSET: _a.append(hours)
+ if mins is not _UNSET: _a.append(mins)
+ if secs is not _UNSET: _a.append(secs)
+ if timezone is not _UNSET: _a.append(timezone)
+ if date is not _UNSET: _a.append(date)
+ if time is not _UNSET: _a.append(time)
+ return _dispatch("try_make_timestamp", _a)
+def try_make_timestamp_ltz(years, months, days, hours, mins, secs, timezone=_UNSET):
+ _a = []
+ _a.append(years)
+ _a.append(months)
+ _a.append(days)
+ _a.append(hours)
+ _a.append(mins)
+ _a.append(secs)
+ if timezone is not _UNSET: _a.append(timezone)
+ return _dispatch("try_make_timestamp_ltz", _a)
+def try_make_timestamp_ntz(years=_UNSET, months=_UNSET, days=_UNSET, hours=_UNSET, mins=_UNSET, secs=_UNSET, date=_UNSET, time=_UNSET):
+ _a = []
+ if years is not _UNSET: _a.append(years)
+ if months is not _UNSET: _a.append(months)
+ if days is not _UNSET: _a.append(days)
+ if hours is not _UNSET: _a.append(hours)
+ if mins is not _UNSET: _a.append(mins)
+ if secs is not _UNSET: _a.append(secs)
+ if date is not _UNSET: _a.append(date)
+ if time is not _UNSET: _a.append(time)
+ return _dispatch("try_make_timestamp_ntz", _a)
+def try_mod(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("try_mod", _a)
+def try_multiply(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("try_multiply", _a)
+def try_parse_json(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("try_parse_json", _a)
+def try_parse_url(url, partToExtract, key=_UNSET):
+ _a = []
+ _a.append(url)
+ _a.append(partToExtract)
+ if key is not _UNSET: _a.append(key)
+ return _dispatch("try_parse_url", _a)
+def try_reflect(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("try_reflect", _a)
+def try_subtract(left, right):
+ _a = []
+ _a.append(left)
+ _a.append(right)
+ return _dispatch("try_subtract", _a)
+def try_sum(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("try_sum", _a)
+def try_to_binary(col, format=_UNSET):
+ _a = []
+ _a.append(col)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("try_to_binary", _a)
+def try_to_date(col, format=_UNSET):
+ _a = []
+ _a.append(col)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("try_to_date", _a)
+def try_to_number(col, format):
+ _a = []
+ _a.append(col)
+ _a.append(format)
+ return _dispatch("try_to_number", _a)
+def try_to_time(str, format=_UNSET):
+ _a = []
+ _a.append(str)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("try_to_time", _a)
+def try_to_timestamp(col, format=_UNSET):
+ _a = []
+ _a.append(col)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("try_to_timestamp", _a)
+def try_url_decode(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("try_url_decode", _a)
+def try_validate_utf8(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("try_validate_utf8", _a)
try_variant_array_append = _create_wrapper("try_variant_array_append")
-try_variant_get = _create_wrapper("try_variant_get")
+def try_variant_get(v, path, targetType):
+ _a = []
+ _a.append(v)
+ _a.append(path)
+ _a.append(targetType)
+ return _dispatch("try_variant_get", _a)
try_variant_insert = _create_wrapper("try_variant_insert")
try_variant_set = _create_wrapper("try_variant_set")
-tuple_difference_double = _create_wrapper("tuple_difference_double")
-tuple_difference_integer = _create_wrapper("tuple_difference_integer")
-tuple_difference_theta_double = _create_wrapper("tuple_difference_theta_double")
-tuple_difference_theta_integer = _create_wrapper("tuple_difference_theta_integer")
-tuple_intersection_agg_double = _create_wrapper("tuple_intersection_agg_double")
-tuple_intersection_agg_integer = _create_wrapper("tuple_intersection_agg_integer")
-tuple_intersection_double = _create_wrapper("tuple_intersection_double")
-tuple_intersection_integer = _create_wrapper("tuple_intersection_integer")
-tuple_intersection_theta_double = _create_wrapper("tuple_intersection_theta_double")
-tuple_intersection_theta_integer = _create_wrapper("tuple_intersection_theta_integer")
-tuple_sketch_agg_double = _create_wrapper("tuple_sketch_agg_double")
-tuple_sketch_agg_integer = _create_wrapper("tuple_sketch_agg_integer")
-tuple_sketch_estimate_double = _create_wrapper("tuple_sketch_estimate_double")
-tuple_sketch_estimate_integer = _create_wrapper("tuple_sketch_estimate_integer")
-tuple_sketch_summary_double = _create_wrapper("tuple_sketch_summary_double")
-tuple_sketch_summary_integer = _create_wrapper("tuple_sketch_summary_integer")
-tuple_sketch_theta_double = _create_wrapper("tuple_sketch_theta_double")
-tuple_sketch_theta_integer = _create_wrapper("tuple_sketch_theta_integer")
-tuple_union_agg_double = _create_wrapper("tuple_union_agg_double")
-tuple_union_agg_integer = _create_wrapper("tuple_union_agg_integer")
-tuple_union_double = _create_wrapper("tuple_union_double")
-tuple_union_integer = _create_wrapper("tuple_union_integer")
-tuple_union_theta_double = _create_wrapper("tuple_union_theta_double")
-tuple_union_theta_integer = _create_wrapper("tuple_union_theta_integer")
-typeof = _create_wrapper("typeof")
-ucase = _create_wrapper("ucase")
-unbase64 = _create_wrapper("unbase64")
-unhex = _create_wrapper("unhex")
-uniform = _create_wrapper("uniform")
-unix_date = _create_wrapper("unix_date")
-unix_micros = _create_wrapper("unix_micros")
-unix_millis = _create_wrapper("unix_millis")
+def tuple_difference_double(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("tuple_difference_double", _a)
+def tuple_difference_integer(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("tuple_difference_integer", _a)
+def tuple_difference_theta_double(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("tuple_difference_theta_double", _a)
+def tuple_difference_theta_integer(col1, col2):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ return _dispatch("tuple_difference_theta_integer", _a)
+def tuple_intersection_agg_double(col, mode=_UNSET):
+ _a = []
+ _a.append(col)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_intersection_agg_double", _a)
+def tuple_intersection_agg_integer(col, mode=_UNSET):
+ _a = []
+ _a.append(col)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_intersection_agg_integer", _a)
+def tuple_intersection_double(col1, col2, mode=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_intersection_double", _a)
+def tuple_intersection_integer(col1, col2, mode=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_intersection_integer", _a)
+def tuple_intersection_theta_double(col1, col2, mode=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_intersection_theta_double", _a)
+def tuple_intersection_theta_integer(col1, col2, mode=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_intersection_theta_integer", _a)
+def tuple_sketch_agg_double(key, summary, lgNomEntries=_UNSET, mode=_UNSET):
+ _a = []
+ _a.append(key)
+ _a.append(summary)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_sketch_agg_double", _a)
+def tuple_sketch_agg_integer(key, summary, lgNomEntries=_UNSET, mode=_UNSET):
+ _a = []
+ _a.append(key)
+ _a.append(summary)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_sketch_agg_integer", _a)
+def tuple_sketch_estimate_double(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("tuple_sketch_estimate_double", _a)
+def tuple_sketch_estimate_integer(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("tuple_sketch_estimate_integer", _a)
+def tuple_sketch_summary_double(col, mode=_UNSET):
+ _a = []
+ _a.append(col)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_sketch_summary_double", _a)
+def tuple_sketch_summary_integer(col, mode=_UNSET):
+ _a = []
+ _a.append(col)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_sketch_summary_integer", _a)
+def tuple_sketch_theta_double(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("tuple_sketch_theta_double", _a)
+def tuple_sketch_theta_integer(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("tuple_sketch_theta_integer", _a)
+def tuple_union_agg_double(col, lgNomEntries=_UNSET, mode=_UNSET):
+ _a = []
+ _a.append(col)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_union_agg_double", _a)
+def tuple_union_agg_integer(col, lgNomEntries=_UNSET, mode=_UNSET):
+ _a = []
+ _a.append(col)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_union_agg_integer", _a)
+def tuple_union_double(col1, col2, lgNomEntries=_UNSET, mode=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_union_double", _a)
+def tuple_union_integer(col1, col2, lgNomEntries=_UNSET, mode=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_union_integer", _a)
+def tuple_union_theta_double(col1, col2, lgNomEntries=_UNSET, mode=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_union_theta_double", _a)
+def tuple_union_theta_integer(col1, col2, lgNomEntries=_UNSET, mode=_UNSET):
+ _a = []
+ _a.append(col1)
+ _a.append(col2)
+ if lgNomEntries is not _UNSET: _a.append(lgNomEntries)
+ if mode is not _UNSET: _a.append(mode)
+ return _dispatch("tuple_union_theta_integer", _a)
+def typeof(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("typeof", _a)
+def ucase(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("ucase", _a)
+def unbase64(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("unbase64", _a)
+def unhex(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("unhex", _a)
+def uniform(min, max, seed=_UNSET):
+ _a = []
+ _a.append(min)
+ _a.append(max)
+ if seed is not _UNSET: _a.append(seed)
+ return _dispatch("uniform", _a)
+def unix_date(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("unix_date", _a)
+def unix_micros(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("unix_micros", _a)
+def unix_millis(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("unix_millis", _a)
unix_nanos = _create_wrapper("unix_nanos")
-unix_seconds = _create_wrapper("unix_seconds")
-unix_timestamp = _create_wrapper("unix_timestamp")
-unwrap_udt = _create_wrapper("unwrap_udt")
-upper = _create_wrapper("upper")
-url_decode = _create_wrapper("url_decode")
-url_encode = _create_wrapper("url_encode")
-user = _create_wrapper("user")
-uuid = _create_wrapper("uuid")
-validate_utf8 = _create_wrapper("validate_utf8")
-var_pop = _create_wrapper("var_pop")
-var_samp = _create_wrapper("var_samp")
-variance = _create_wrapper("variance")
+def unix_seconds(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("unix_seconds", _a)
+def unix_timestamp(timestamp=_UNSET, format=_UNSET):
+ _a = []
+ if timestamp is not _UNSET: _a.append(timestamp)
+ if format is not _UNSET: _a.append(format)
+ return _dispatch("unix_timestamp", _a)
+def unwrap_udt(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("unwrap_udt", _a)
+def upper(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("upper", _a)
+def url_decode(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("url_decode", _a)
+def url_encode(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("url_encode", _a)
+def user():
+ _a = []
+ return _dispatch("user", _a)
+def uuid(seed=_UNSET):
+ _a = []
+ if seed is not _UNSET: _a.append(seed)
+ return _dispatch("uuid", _a)
+def validate_utf8(str):
+ _a = []
+ _a.append(str)
+ return _dispatch("validate_utf8", _a)
+def var_pop(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("var_pop", _a)
+def var_samp(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("var_samp", _a)
+def variance(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("variance", _a)
variant_array_append = _create_wrapper("variant_array_append")
-variant_get = _create_wrapper("variant_get")
+def variant_get(v, path, targetType):
+ _a = []
+ _a.append(v)
+ _a.append(path)
+ _a.append(targetType)
+ return _dispatch("variant_get", _a)
variant_insert = _create_wrapper("variant_insert")
variant_set = _create_wrapper("variant_set")
vector_avg = _create_wrapper("vector_avg")
@@ -515,22 +2351,91 @@ vector_l2_distance = _create_wrapper("vector_l2_distance")
vector_norm = _create_wrapper("vector_norm")
vector_normalize = _create_wrapper("vector_normalize")
vector_sum = _create_wrapper("vector_sum")
-version = _create_wrapper("version")
-weekday = _create_wrapper("weekday")
-weekofyear = _create_wrapper("weekofyear")
-when = _create_wrapper("when")
-width_bucket = _create_wrapper("width_bucket")
-window_time = _create_wrapper("window_time")
-xpath = _create_wrapper("xpath")
-xpath_boolean = _create_wrapper("xpath_boolean")
-xpath_double = _create_wrapper("xpath_double")
-xpath_float = _create_wrapper("xpath_float")
-xpath_int = _create_wrapper("xpath_int")
-xpath_long = _create_wrapper("xpath_long")
-xpath_number = _create_wrapper("xpath_number")
-xpath_short = _create_wrapper("xpath_short")
-xpath_string = _create_wrapper("xpath_string")
-xxhash64 = _create_wrapper("xxhash64")
-year = _create_wrapper("year")
-years = _create_wrapper("years")
-zeroifnull = _create_wrapper("zeroifnull")
+def version():
+ _a = []
+ return _dispatch("version", _a)
+def weekday(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("weekday", _a)
+def weekofyear(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("weekofyear", _a)
+def when(condition, value):
+ _a = []
+ _a.append(condition)
+ _a.append(value)
+ return _dispatch("when", _a)
+def width_bucket(v, min, max, numBucket):
+ _a = []
+ _a.append(v)
+ _a.append(min)
+ _a.append(max)
+ _a.append(numBucket)
+ return _dispatch("width_bucket", _a)
+def window_time(windowColumn):
+ _a = []
+ _a.append(windowColumn)
+ return _dispatch("window_time", _a)
+def xpath(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath", _a)
+def xpath_boolean(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath_boolean", _a)
+def xpath_double(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath_double", _a)
+def xpath_float(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath_float", _a)
+def xpath_int(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath_int", _a)
+def xpath_long(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath_long", _a)
+def xpath_number(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath_number", _a)
+def xpath_short(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath_short", _a)
+def xpath_string(xml, path):
+ _a = []
+ _a.append(xml)
+ _a.append(path)
+ return _dispatch("xpath_string", _a)
+def xxhash64(*cols):
+ _a = []
+ _a.extend(cols)
+ return _dispatch("xxhash64", _a)
+def year(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("year", _a)
+def years(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("years", _a)
+def zeroifnull(col):
+ _a = []
+ _a.append(col)
+ return _dispatch("zeroifnull", _a)
diff --git a/python/tests/test_dropin_offline.py b/python/tests/test_dropin_offline.py
index 8b78005..dc01a95 100644
--- a/python/tests/test_dropin_offline.py
+++ b/python/tests/test_dropin_offline.py
@@ -193,6 +193,46 @@ def test_column_methods_offline():
assert type(c.transform(lambda x: x + 1)).__name__ == "Column"
+def test_functions_keyword_and_dispatch_paths():
+ # PySpark 4.2.0 keyword-argument parity: generated wrappers carry the reference
+ # parameter names, and the hand-written wrappers dispatch to the core. All lazy
+ # (column-expression building; no server needed).
+ from pyspark.sql import functions as F
+
+ def is_col(x):
+ return type(x).__name__ == "Column"
+
+ c = F.col("id")
+ # hand-written wrappers with reference parameter names
+ assert is_col(F.col("id")) and is_col(F.lit(3))
+ assert is_col(F.expr("id + 1")) and is_col(F.column("id"))
+ assert is_col(F.sha2(c, 256))
+ assert is_col(F.window(c, "10 minutes"))
+ assert is_col(F.window(c, "10 minutes", "5 minutes", "1 minute"))
+ # generated wrappers accept the reference keyword arguments (GAP 1)
+ assert is_col(F.first(c, ignorenulls=True))
+ assert is_col(F.round(c, scale=2))
+ assert is_col(F.split(c, pattern=",", limit=2))
+ assert is_col(F.array_contains(c, value=1))
+ assert is_col(F.concat_ws("-", c, c))
+ # lag/lead literal-arg overrides
+ assert is_col(F.lag(c, offset=2, default=0))
+ assert is_col(F.lead(c, offset=2, default=0))
+ # the *args fallback wrapper (used for Rust-only extras) still dispatches positionally
+ assert is_col(F._create_wrapper("upper")(c))
+ # the _UNSET sentinel (default for optional generated args) has a stable repr
+ assert repr(F._UNSET) == "<unset>"
+
+
+def test_column_alias_metadata_offline():
+ from pyspark.sql import functions as F
+
+ c = F.col("id")
+ assert type(c.alias("x")).__name__ == "Column"
+ assert type(c.alias("x", metadata={"k": "v"})).__name__ == "Column"
+ assert type(c.name("y")).__name__ == "Column"
+
+
def test_new_v420_methods_exposed():
# Presence of the v4.2.0 parity methods on the drop-in classes (behaviour that
# needs a live server is covered by the e2e tests).
diff --git a/scripts/e2e_parity_check.py b/scripts/e2e_parity_check.py
new file mode 100644
index 0000000..9f14961
--- /dev/null
+++ b/scripts/e2e_parity_check.py
@@ -0,0 +1,126 @@
+"""Focused e2e validation of the parity fixes against a real Spark Connect server.
+Exercises keyword-argument calls and newly-wired params end-to-end (real execution),
+checking results — run with our Rust client on PYTHONPATH."""
+
+import os
+import tempfile
+
+from pyspark.sql import SparkSession
+from pyspark.sql import functions as F
+
+spark = SparkSession.builder.remote(
+ os.environ.get("SPARK_CONNECT_TESTING_REMOTE", "sc://localhost:15002")
+).getOrCreate()
+
+ok = 0
+fail = 0
+
+
+def check(desc, got, want):
+ global ok, fail
+ if got == want:
+ ok += 1
+ print(f"OK {desc} -> {got}")
+ else:
+ fail += 1
+ print(f"FAIL {desc} got={got!r} want={want!r}")
+
+
+def run(desc, fn):
+ global ok, fail
+ try:
+ fn()
+ ok += 1
+ print(f"OK {desc}")
+ except Exception as e:
+ fail += 1
+ print(f"FAIL {desc} -> {type(e).__name__}: {str(e)[:80]}")
+
+
+df = spark.range(10).toDF("id")
+
+# GAP 1: keyword-argument function calls now execute and produce correct plans/results
+check(
+ "F.round(lit(3.14159), scale=2)",
+ spark.range(1).select(F.round(F.lit(3.14159), scale=2).alias("r")).collect()[0]["r"],
+ 3.14,
+)
+check(
+ "F.split(lit('a,b,c'), pattern=',') length",
+ len(spark.range(1).select(F.split(F.lit("a,b,c"), pattern=",").alias("s")).collect()[0]["s"]),
+ 3,
+)
+check(
+ "F.first(id, ignorenulls=True) over range",
+ spark.range(5).select(F.first("id", ignorenulls=True).alias("f")).collect()[0]["f"],
+ 0,
+)
+run(
+ "F.from_json(col, schema='a INT') builds",
+ lambda: (
+ spark.range(1).select(F.from_json(F.lit('{"a":1}'), schema="a INT").alias("j")).collect()
+ ),
+)
+check(
+ "F.concat_ws(sep, *cols)",
+ spark.range(1).select(F.concat_ws("-", F.lit("a"), F.lit("b")).alias("c")).collect()[0]["c"],
+ "a-b",
+)
+
+# GAP 2: method keyword/params
+check(
+ "df.sample(withReplacement=False, fraction=1.0, seed=1) count",
+ spark.range(100).sample(withReplacement=False, fraction=1.0, seed=1).count() > 0,
+ True,
+)
+check(
+ "df.sort(id, ascending=False) first",
+ spark.range(5).sort("id", ascending=False).collect()[0]["id"],
+ 4,
+)
+check(
+ "df.orderBy(id, ascending=[False]) first",
+ spark.range(5).orderBy("id", ascending=[False]).collect()[0]["id"],
+ 4,
+)
+check(
+ "df.selectExpr('id * 2 as d') first",
+ spark.range(3).selectExpr("id * 2 as d").collect()[1]["d"],
+ 2,
+)
+check("df.drop('id') columns", spark.range(3).toDF("id").drop("id").columns, [])
+check(
+ "df.toDF('a','b') columns",
+ spark.range(1).select(F.lit(1), F.lit(2)).toDF("a", "b").columns,
+ ["a", "b"],
+)
+run(
+ "df.unpivot(ids, values, variableColumnName=, valueColumnName=)",
+ lambda: (
+ spark.range(1)
+ .select(F.lit(1).alias("a"), F.lit(2).alias("b"))
+ .unpivot(["a"], ["b"], variableColumnName="var", valueColumnName="val")
+ .collect()
+ ),
+)
+check(
+ "spark.createDataFrame(data, schema, verifySchema=True)",
+ spark.createDataFrame([(1, "x")], "a int, b string", verifySchema=True).count(),
+ 1,
+)
+check("spark.sql(':p', p=...) via kwargs", spark.sql("SELECT :p AS v", p=7).collect()[0]["v"], 7)
+
+# write/read round-trip exercising save(format=,mode=) + read.load(format=)
+d = tempfile.mkdtemp()
+p = os.path.join(d, "t")
+run(
+ "df.write.save(path, format='parquet', mode='overwrite')",
+ lambda: spark.range(5).toDF("id").write.save(p, format="parquet", mode="overwrite"),
+)
+check(
+ "spark.read.load(path, format='parquet') count", spark.read.load(p, format="parquet").count(), 5
+)
+
+print(f"\n==== e2e parity: {ok} OK, {fail} FAIL ====")
+spark.stop()
+raise SystemExit(1 if fail else 0)
diff --git a/scripts/gen_fn_dispatch.py b/scripts/gen_fn_dispatch.py
index a708c19..1bf5539 100644
--- a/scripts/gen_fn_dispatch.py
+++ b/scripts/gen_fn_dispatch.py
@@ -4,9 +4,36 @@ Generator for dispatch and Python wrappers for all 440 Spark SQL functions.
Parses crates/spark-connect/src/functions.rs and generates optimized code.
"""
+import argparse
+import inspect
import re
+import sys
from pathlib import Path
+_ap = argparse.ArgumentParser(description=__doc__)
+_ap.add_argument(
+ "--pyspark-src",
+ default=None,
+ help="Path to the reference PySpark package root (…/python). When given, generated "
+ "Python wrappers carry the reference's real parameter names/order so keyword calls "
+ "work; without it, wrappers fall back to the historical *args form.",
+)
+_ap.add_argument(
+ "--write-dispatch",
+ action="store_true",
+ help="Also (re)write crates/pyspark-rs/src/dispatch_generated.rs. OFF by default: "
+ "that file carries hand-tuned multi-arg arms (e.g. round_scale, "
+ "approx_count_distinct_rsd) this generator does not reproduce, so overwriting it "
+ "would regress. Only pass this when intentionally regenerating the Rust dispatch.",
+)
+_args = _ap.parse_args()
+
+# Optionally load the reference pyspark to mirror real function signatures.
+_ref_functions = None
+if _args.pyspark_src:
+ sys.path.insert(0, _args.pyspark_src)
+ import pyspark.sql.functions as _ref_functions # noqa: E402
+
# Read the spark-connect functions file
functions_rs = Path("crates/spark-connect/src/functions.rs").read_text()
@@ -238,17 +265,95 @@ pub fn call_builtin(name: &str, args: Vec<spark_connect::column::Column>) -> PyR
"""
)
-# Write dispatch file
+# Write dispatch file (guarded: the committed file is hand-tuned; see --write-dispatch).
dispatch_file = Path("crates/pyspark-rs/src/dispatch_generated.rs")
-dispatch_file.write_text(dispatch_code)
-print(f"\nGenerated dispatch function in {dispatch_file}")
+if _args.write_dispatch:
+ dispatch_file.write_text(dispatch_code)
+ print(f"\nGenerated dispatch function in {dispatch_file}")
+else:
+ print(f"\nSkipped {dispatch_file} (pass --write-dispatch to regenerate the Rust dispatch)")
# Generate Python wrapper functions
# (skip_functions is defined above, before categorization.)
+#
+# When a reference PySpark is available, emit an explicit ``def`` per function carrying
+# the reference's real parameter names/order, so keyword calls (``F.first(col,
+# ignorenulls=True)``) work. Optional params default to the ``_UNSET`` sentinel and are
+# forwarded only when supplied, reproducing the historical ``*args`` dispatch exactly
+# (same columns, same order) — see functions.py ``_dispatch`` / ``_UNSET``.
+
+
+def _emit_def(name, sig):
+ """Return source for ``def name(<ref params>): return _dispatch(name, [...])``."""
+ sig_parts = []
+ build = []
+ posonly = []
+ star_emitted = False
+ for p in sig.parameters.values():
+ k = p.kind
+ if k is inspect.Parameter.VAR_KEYWORD:
+ continue # **kwargs are not part of the positional dispatch
+ if k is inspect.Parameter.POSITIONAL_ONLY:
+ posonly.append(p.name)
+ sig_parts.append(p.name)
+ build.append(f" _a.append({p.name})")
+ elif k is inspect.Parameter.POSITIONAL_OR_KEYWORD:
+ if p.default is inspect.Parameter.empty:
+ sig_parts.append(p.name)
+ build.append(f" _a.append({p.name})")
+ else:
+ sig_parts.append(f"{p.name}=_UNSET")
+ build.append(f" if {p.name} is not _UNSET: _a.append({p.name})")
+ elif k is inspect.Parameter.VAR_POSITIONAL:
+ sig_parts.append(f"*{p.name}")
+ build.append(f" _a.extend({p.name})")
+ star_emitted = True
+ elif k is inspect.Parameter.KEYWORD_ONLY:
+ if not star_emitted:
+ sig_parts.append("*")
+ star_emitted = True
+ sig_parts.append(f"{p.name}=_UNSET")
+ build.append(f" if {p.name} is not _UNSET: _a.append({p.name})")
+ if posonly:
+ # insert the positional-only marker after the last positional-only param
+ idx = len(posonly)
+ sig_parts.insert(idx, "/")
+ header = f"def {name}({', '.join(sig_parts)}):"
+ body = [" _a = []", *build, f' return _dispatch("{name}", _a)']
+ return header + "\n" + "\n".join(body)
+
+
+# The authoritative wrapper set is the names already in functions_generated.py (the
+# shipped, golden-verified set). Deriving it from the Rust functions.rs regex is fragile:
+# generic signatures (e.g. `pub fn concat_ws<C: Into<Column>>(...)`) are silently missed.
+# We only *upgrade* each existing wrapper's signature; we never add or drop functions.
+wrappers_file = Path("python/pyspark/sql/functions_generated.py")
+_existing = wrappers_file.read_text() if wrappers_file.exists() else ""
+_wrapper_names = re.findall(r"^([A-Za-z_][A-Za-z0-9_]*)\s*=\s*_create_wrapper\(", _existing, re.M)
+_wrapper_names += re.findall(r"^def ([A-Za-z_][A-Za-z0-9_]*)\(", _existing, re.M)
+_wrapper_names = sorted(set(_wrapper_names))
+if not _wrapper_names:
+ # First-time generation: fall back to the Rust-parsed function set.
+ _wrapper_names = sorted(n for n in functions if n not in skip_functions)
+
python_wrappers = []
-for name in sorted(functions.keys()):
- if name not in skip_functions:
- python_wrappers.append(f'''{name} = _create_wrapper("{name}")''')
+n_typed = 0
+for name in _wrapper_names:
+ if name in skip_functions:
+ continue
+ ref_fn = getattr(_ref_functions, name, None) if _ref_functions is not None else None
+ sig = None
+ if ref_fn is not None and callable(ref_fn):
+ try:
+ sig = inspect.signature(ref_fn)
+ except (TypeError, ValueError):
+ sig = None
+ if sig is not None:
+ python_wrappers.append(_emit_def(name, sig))
+ n_typed += 1
+ else:
+ # Rust-only extra (not in the reference) — keep the historical *args form.
+ python_wrappers.append(f'{name} = _create_wrapper("{name}")')
python_code = (
'''"""Auto-generated wrapper functions for Spark SQL functions.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org