diff --git a/python/pyspark/sql/connect/readwriter.py b/python/pyspark/sql/connect/readwriter.py index 561447a437e6d..6da2982e92c40 100644 --- a/python/pyspark/sql/connect/readwriter.py +++ b/python/pyspark/sql/connect/readwriter.py @@ -111,6 +111,11 @@ def schema(self, schema: Union[StructType, str]) -> "DataFrameReader": schema.__doc__ = PySparkDataFrameReader.schema.__doc__ def option(self, key: str, value: "OptionalPrimitiveType") -> "DataFrameReader": + # Remove case-insensitive matches so the latest spelling and value win. + normalized_key = key.lower() + for existing_key in list(self._options): + if existing_key.lower() == normalized_key: + del self._options[existing_key] self._options[key] = cast(str, to_str(value)) return self diff --git a/python/pyspark/sql/tests/connect/test_connect_plan.py b/python/pyspark/sql/tests/connect/test_connect_plan.py index 685ea3656f5cb..6069b703b72f2 100644 --- a/python/pyspark/sql/tests/connect/test_connect_plan.py +++ b/python/pyspark/sql/tests/connect/test_connect_plan.py @@ -772,6 +772,22 @@ def test_datasource_read(self): self.assertEqual(len(data_source.paths), 1) self.assertEqual(data_source.paths[0], "test_path") + def test_reader_options_case_insensitive(self): + reader = DataFrameReader(self.connect) + df = ( + reader.option("versionAsOf", 0) + .option("versionasof", 1) + .option("versionAsOf", 2) + .table("myTable") + ) + options = df._plan.to_proto(self.connect).root.read.named_table.options + self.assertEqual(dict(options), {"versionAsOf": "2"}) + + reader = DataFrameReader(self.connect) + df = reader.option("header", False).option("HEADER", True).load(format="csv") + options = df._plan.to_proto(self.connect).root.read.data_source.options + self.assertEqual(dict(options), {"HEADER": "true"}) + def test_relation_changes(self): reader = DataFrameReader(self.connect) df = reader.option("startingVersion", "1").option("endingVersion", "5").changes("myTable")