robin-sparkless 4.4.0

PySpark-like DataFrame API in Rust on Polars; no JVM.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173

"""
Tests for issue #360: input_file_name() function.

PySpark supports F.input_file_name() which returns the path of the file being read.
Sparkless returns empty string in mock (no file path context).

Works with both sparkless (mock) and PySpark backends.
Set SPARKLESS_TEST_MODE=pyspark to run with real PySpark.
"""


class TestIssue360InputFileName:
    """Test input_file_name() (issue #360)."""

    def test_input_file_name_returns_string_column(self, spark):
        """F.input_file_name() returns a string column (empty in mock)."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame(
            [
                {"dataset": "dataset_a", "table": "table_1"},
                {"dataset": "dataset_b", "table": "table_2"},
            ]
        )
        result = df.withColumn("InputFileName", F.input_file_name())
        rows = result.collect()
        assert len(rows) == 2
        assert "InputFileName" in rows[0]
        # In Sparkless mock we return empty string; PySpark returns actual path
        assert isinstance(rows[0]["InputFileName"], str)

    def test_input_file_name_exact_issue_scenario(self, spark):
        """Exact scenario from issue #360: withColumn after createDataFrame."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame(
            [
                ("dataset_a", "table_1"),
                ("dataset_b", "table_2"),
            ],
            ["dataset", "table"],
        )
        df = df.withColumn("InputFileName", F.input_file_name())
        rows = df.collect()
        assert len(rows) == 2
        assert rows[0]["dataset"] == "dataset_a"
        assert rows[0]["table"] == "table_1"
        assert "InputFileName" in rows[0]

    def test_input_file_name_select_only(self, spark):
        """Select only input_file_name() as single column."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"a": 1}, {"a": 2}])
        result = df.select(F.input_file_name().alias("path"))
        rows = result.collect()
        assert len(rows) == 2
        assert (
            rows[0]["path"] == ""
            or rows[0]["path"].startswith("/")
            or "\\" in rows[0]["path"]
        )


class TestIssue360InputFileNameRobust:
    """Robust tests for input_file_name(): edge cases, chaining, schema."""

    def test_input_file_name_empty_dataframe(self, spark):
        """input_file_name() on empty DataFrame returns empty result."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([], "a int")
        result = df.withColumn("path", F.input_file_name())
        rows = result.collect()
        assert len(rows) == 0

    def test_input_file_name_single_row(self, spark):
        """input_file_name() with single row."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"id": 1}])
        result = df.withColumn("file", F.input_file_name())
        rows = result.collect()
        assert len(rows) == 1
        assert "file" in rows[0]
        assert isinstance(rows[0]["file"], str)

    def test_input_file_name_after_filter(self, spark):
        """input_file_name() after filter."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"a": 1}, {"a": 2}, {"a": 3}])
        result = df.filter(F.col("a") > 1).withColumn("path", F.input_file_name())
        rows = result.collect()
        assert len(rows) == 2
        assert all("path" in r and isinstance(r["path"], str) for r in rows)

    def test_input_file_name_after_select(self, spark):
        """input_file_name() after select (different columns)."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"x": 1, "y": 2}, {"x": 3, "y": 4}])
        result = df.select("x").withColumn("path", F.input_file_name())
        rows = result.collect()
        assert len(rows) == 2
        assert rows[0]["path"] == "" or isinstance(rows[0]["path"], str)

    def test_input_file_name_preserves_schema(self, spark):
        """withColumn(input_file_name()) adds string column; schema preserved."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"a": 1, "b": "x"}])
        result = df.withColumn("path", F.input_file_name())
        assert "path" in result.schema.fieldNames()
        assert len(result.schema.fields) == 3
        rows = result.collect()
        assert rows[0]["a"] == 1 and rows[0]["b"] == "x" and "path" in rows[0]

    def test_input_file_name_with_show(self, spark):
        """withColumn(input_file_name()).show() does not raise."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"a": 1}])
        result = df.withColumn("path", F.input_file_name())
        result.show()

    def test_input_file_name_all_rows_same_type(self, spark):
        """All rows get a string value from input_file_name() (no None in mock)."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"i": i} for i in range(5)])
        result = df.withColumn("path", F.input_file_name())
        rows = result.collect()
        for r in rows:
            assert "path" in r
            assert isinstance(r["path"], str)

    def test_input_file_name_multiple_columns(self, spark):
        """input_file_name() alongside other columns in select."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"a": 1, "b": 2}, {"a": 3, "b": 4}])
        result = df.select(
            F.col("a"),
            F.input_file_name().alias("path"),
            F.col("b"),
        )
        rows = result.collect()
        assert len(rows) == 2
        assert rows[0]["a"] == 1 and rows[0]["b"] == 2 and "path" in rows[0]

    def test_input_file_name_alias(self, spark):
        """input_file_name() with alias."""
        from sparkless.testing import get_imports

        F = get_imports().F
        df = spark.createDataFrame([{"x": 1}])
        result = df.select(F.input_file_name().alias("source_file"))
        rows = result.collect()
        assert len(rows) == 1
        assert "source_file" in rows[0]
        assert isinstance(rows[0]["source_file"], str)