[python] Fix wrong result when using where and limit in CLI table read

Author: JingsongLi

paimon-python/pypaimon/cli/cli_table.py

@@ -104,9 +104,11 @@ def cmd_table_read(args):
             print(f"Error: Invalid WHERE clause: {e}", file=sys.stderr)
             sys.exit(1)
 
-    # Apply limit if specified
+    # Apply limit: only push down when there is no where clause,
+    # because limit push-down may stop reading before enough rows
+    # pass the filter, leading to fewer results than expected.
     limit = args.limit
-    if limit:
+    if limit and not where_clause:
         read_builder = read_builder.with_limit(limit)
 
     # Scan and read

paimon-python/pypaimon/tests/cli_table_test.py

@@ -1049,6 +1049,60 @@ def test_cli_table_read_where_field_not_in_select(self):
                 self.assertNotIn(' 32', output)
                 self.assertNotIn(' 25', output)
 
+    def test_cli_table_read_with_where_and_limit(self):
+        """Test that where + limit returns correct filtered results without limit push-down.
+
+        Writes data in two batches to produce multiple splits, so that limit
+        push-down would actually take effect and potentially miss matching rows
+        in later splits.
+        """
+
+        # Create a dedicated table for this test with two batches of data
+        pa_schema = pa.schema([
+            ('id', pa.int32()),
+            ('val', pa.string()),
+            ('score', pa.int32()),
+        ])
+        # Important: multiple splits are required for the limit to take effect
+        schema = Schema.from_pyarrow_schema(pa_schema, options={'source.split.target-size': '1b'})
+        self.catalog.create_table('test_db.limit_test', schema, True)
+        table = self.catalog.get_table('test_db.limit_test')
+
+        def write_batch():
+            write_builder = table.new_batch_write_builder()
+            table_write = write_builder.new_write()
+            table_commit = write_builder.new_commit()
+            batch = pa.Table.from_pydict({
+                'id': [1, 2, 3],
+                'val': ['a', 'b', 'c'],
+                'score': [10, 20, 30],
+            }, schema=pa_schema)
+            table_write.write_arrow(batch)
+            table_commit.commit(table_write.prepare_commit())
+            table_write.close()
+            table_commit.close()
+
+        write_batch()
+        write_batch()
+        write_batch()
+
+        with patch('sys.argv',
+                   ['paimon', '-c', self.config_file,
+                    'table', 'read', 'test_db.limit_test',
+                    '--where', 'score = 20',
+                    '--limit', '2']):
+            with patch('sys.stdout', new_callable=StringIO) as mock_stdout:
+                try:
+                    main()
+                except SystemExit:
+                    pass
+
+                output = mock_stdout.getvalue()
+                lines = [line for line in output.strip().split('\n') if line.strip()]
+                self.assertEqual(len(lines), 3)
+                self.assertNotIn(' a ', output)
+                self.assertNotIn(' c ', output)
+
     def test_cli_table_read_with_invalid_where(self):
         """Test table read with invalid --where clause via CLI."""
         with patch('sys.argv',