diff --git a/python/pyspark/sql/column.py b/python/pyspark/sql/column.py
index 1ecec5b126505..6ce707ac57d09 100644
--- a/python/pyspark/sql/column.py
+++ b/python/pyspark/sql/column.py
@@ -396,6 +396,11 @@ def over(self, window):
         jc = self._jc.over(window._jspec)
         return Column(jc)
 
+    def __nonzero__(self):
+        raise ValueError("Can't convert column into bool: please use '&' for 'and', '|' for 'or', "
+                         "when using Column in a boolean expression.")
+    __bool__ = __nonzero__
+
     def __repr__(self):
         return 'Column<%s>' % self._jc.toString().encode('utf8')
 
diff --git a/python/pyspark/sql/tests.py b/python/pyspark/sql/tests.py
index 13f4556943ac8..e641776667185 100644
--- a/python/pyspark/sql/tests.py
+++ b/python/pyspark/sql/tests.py
@@ -164,6 +164,13 @@ def test_explode(self):
         self.assertEqual(result[0][0], "a")
         self.assertEqual(result[0][1], "b")
 
+    def test_and_in_expression(self):
+        self.assertEqual(4, self.df.filter(self.df.key <= 10 & self.df.value <= "2").count())
+        self.assertRaises(ValueError, lambda: self.df.key <= 10 & self.df.value <= "2")
+        self.assertEqual(2, self.df.filter(self.df.key <= 3 | self.df.value < "2").count())
+        self.assertRaises(ValueError,
+                          lambda: self.df.filter(self.df.key <= 3 | self.df.value < "2").count())
+
     def test_udf_with_callable(self):
         d = [Row(number=i, squared=i**2) for i in range(10)]
         rdd = self.sc.parallelize(d)